AWS EC2成本优化完全指南：深度解析EC2-Other费用并提供技术优化方案

典型EC2-Other月度账单分解（$1000示例）：
- EBS存储卷: $400 (40%)
  - gp3: $200
  - gp2: $100
  - io1/io2: $100
- EBS快照: $100 (10%)
- 数据传输: $250 (25%)
  - 跨区域: $100
  - Internet出站: $150
- NAT网关: $150 (15%)
  - 固定费用: $45
  - 数据处理: $105
- 弹性IP: $50 (5%)
- 负载均衡器: $50 (5%)

import boto3
import time

ec2 = boto3.client('ec2')

def optimize_ebs_volumes():
    """AWS EC2成本优化 - 自动转换gp2到gp3"""

    # 获取所有gp2卷
    volumes = ec2.describe_volumes(
        Filters=[
            {'Name': 'volume-type', 'Values': ['gp2']},
            {'Name': 'status', 'Values': ['in-use', 'available']}
        ]
    )

    total_savings = 0

    for volume in volumes['Volumes']:
        volume_id = volume['VolumeId']
        size = volume['Size']

        # 计算潜在节省
        gp2_cost = size * 0.10  # $0.10 per GB for gp2
        gp3_cost = size * 0.08  # $0.08 per GB for gp3
        savings = gp2_cost - gp3_cost

        print(f"Converting {volume_id} from gp2 to gp3...")

        # 创建快照（安全备份）
        snapshot = ec2.create_snapshot(
            VolumeId=volume_id,
            Description=f'Backup before gp2 to gp3 conversion'
        )

        # 等待快照完成
        waiter = ec2.get_waiter('snapshot_completed')
        waiter.wait(SnapshotIds=[snapshot['SnapshotId']])

        # 修改卷类型
        ec2.modify_volume(
            VolumeId=volume_id,
            VolumeType='gp3',
            Iops=3000,  # gp3 baseline
            Throughput=125  # gp3 baseline MB/s
        )

        total_savings += savings
        print(f"Converted {volume_id}. Monthly savings: ${savings:.2f}")

    print(f"nTotal monthly savings: ${total_savings:.2f}")
    print(f"Annual savings: ${total_savings * 12:.2f}")

# 执行优化
optimize_ebs_volumes()

def setup_snapshot_lifecycle():
    """AWS EC2成本优化 - 自动化快照管理"""

    dlm = boto3.client('dlm')

    # 创建生命周期策略
    policy = {
        'PolicyType': 'EBS_SNAPSHOT_MANAGEMENT',
        'ResourceTypes': ['VOLUME'],
        'TargetTags': [
            {'Key': 'Environment', 'Value': 'Production'}
        ],
        'Schedules': [
            {
                'Name': 'Daily Snapshots',
                'CreateRule': {
                    'Interval': 24,
                    'IntervalUnit': 'HOURS',
                    'Times': ['03:00']
                },
                'RetainRule': {
                    'Count': 7  # 保留7天
                },
                'CopyTags': True
            },
            {
                'Name': 'Weekly Snapshots',
                'CreateRule': {
                    'CronExpression': 'cron(0 3 ? * 1 *)'  # 每周一
                },
                'RetainRule': {
                    'Count': 4  # 保留4周
                }
            }
        ]
    }

    response = dlm.create_lifecycle_policy(
        ExecutionRoleArn='arn:aws:iam::account:role/AWSDataLifecycleManagerRole',
        Description='Optimized snapshot policy for cost reduction',
        State='ENABLED',
        PolicyDetails=policy
    )

    print(f"Lifecycle policy created: {response['PolicyId']}")

# 清理旧快照
def cleanup_old_snapshots(days_to_keep=30):
    """删除超过指定天数的快照"""

    ec2 = boto3.client('ec2')
    account_id = boto3.client('sts').get_caller_identity()['Account']

    snapshots = ec2.describe_snapshots(
        OwnerIds=[account_id],
        Filters=[
            {'Name': 'status', 'Values': ['completed']}
        ]
    )

    cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_to_keep)
    deleted_count = 0
    saved_gb = 0

    for snapshot in snapshots['Snapshots']:
        if snapshot['StartTime'] < cutoff_date:
            try:
                ec2.delete_snapshot(SnapshotId=snapshot['SnapshotId'])
                deleted_count += 1
                saved_gb += snapshot.get('VolumeSize', 0)
            except Exception as e:
                print(f"Error deleting {snapshot['SnapshotId']}: {e}")

    monthly_savings = saved_gb * 0.05  # $0.05 per GB-month
    print(f"Deleted {deleted_count} snapshots")
    print(f"Freed {saved_gb} GB")
    print(f"Monthly savings: ${monthly_savings:.2f}")

def create_vpc_endpoints():
    """AWS EC2成本优化 - 创建VPC端点避免数据传输费"""

    ec2 = boto3.client('ec2')

    # 获取VPC信息
    vpcs = ec2.describe_vpcs()

    for vpc in vpcs['Vpcs']:
        vpc_id = vpc['VpcId']

        # 创建S3端点（网关端点，免费）
        try:
            s3_endpoint = ec2.create_vpc_endpoint(
                VpcId=vpc_id,
                ServiceName='com.amazonaws.region.s3',
                RouteTableIds=get_route_tables(vpc_id)
            )
            print(f"Created S3 endpoint for VPC {vpc_id}")
        except Exception as e:
            print(f"S3 endpoint may already exist: {e}")

        # 创建其他服务端点
        services = [
            'com.amazonaws.region.ec2',
            'com.amazonaws.region.rds',
            'com.amazonaws.region.lambda'
        ]

        for service in services:
            try:
                endpoint = ec2.create_vpc_endpoint(
                    VpcId=vpc_id,
                    ServiceName=service,
                    VpcEndpointType='Interface',
                    SubnetIds=get_private_subnets(vpc_id),
                    SecurityGroupIds=get_security_groups(vpc_id)
                )
                print(f"Created {service} endpoint for VPC {vpc_id}")
            except Exception as e:
                print(f"Error creating {service} endpoint: {e}")

def optimize_cross_az_traffic():
    """将相关资源迁移到同一可用区"""

    ec2 = boto3.client('ec2')
    elb = boto3.client('elbv2')

    # 分析当前流量模式
    instances = ec2.describe_instances(
        Filters=[
            {'Name': 'instance-state-name', 'Values': ['running']}
        ]
    )

    az_distribution = {}

    for reservation in instances['Reservations']:
        for instance in reservation['Instances']:
            az = instance['Placement']['AvailabilityZone']
            if az not in az_distribution:
                az_distribution[az] = []
            az_distribution[az].append(instance['InstanceId'])

    # 找出最优AZ（实例最多的）
    optimal_az = max(az_distribution, key=lambda x: len(az_distribution[x]))

    print(f"Optimal AZ: {optimal_az}")
    print(f"Current distribution: {az_distribution}")

    # 计算迁移后的节省
    cross_az_gb = estimate_cross_az_traffic()  # 自定义函数
    monthly_savings = cross_az_gb * 0.01

    print(f"Estimated monthly savings: ${monthly_savings:.2f}")

    return optimal_az, az_distribution

def deploy_nat_instance():
    """AWS EC2成本优化 - 部署NAT实例替代NAT网关"""

    ec2 = boto3.client('ec2')

    # 查找最新的Amazon Linux 2 AMI
    ami_response = ec2.describe_images(
        Owners=['amazon'],
        Filters=[
            {'Name': 'name', 'Values': ['amzn2-ami-hvm-*']},
            {'Name': 'architecture', 'Values': ['x86_64']}
        ]
    )

    latest_ami = sorted(
        ami_response['Images'],
        key=lambda x: x['CreationDate'],
        reverse=True
    )[0]['ImageId']

    # NAT实例用户数据脚本
    user_data = '''#!/bin/bash
    yum update -y
    echo "net.ipv4.ip_forward = 1" >> /etc/sysctl.conf
    sysctl -p

    # 配置iptables
    iptables -t nat -A POSTROUTING -o eth0 -j MASQUERADE
    iptables-save > /etc/sysconfig/iptables

    # 禁用源/目标检查
    TOKEN=`curl -X PUT "http://169.254.169.254/latest/api/token" -H "X-aws-ec2-metadata-token-ttl-seconds: 21600"`
    INSTANCE_ID=`curl -H "X-aws-ec2-metadata-token: $TOKEN" http://169.254.169.254/latest/meta-data/instance-id`
    aws ec2 modify-instance-attribute --instance-id $INSTANCE_ID --no-source-dest-check --region us-east-1
    '''

    # 启动NAT实例
    response = ec2.run_instances(
        ImageId=latest_ami,
        InstanceType='t3.medium',  # 可根据流量调整
        MinCount=1,
        MaxCount=1,
        KeyName='your-key-pair',
        SecurityGroupIds=['sg-nat-instance'],
        SubnetId='subnet-public',
        UserData=user_data,
        TagSpecifications=[
            {
                'ResourceType': 'instance',
                'Tags': [
                    {'Key': 'Name', 'Value': 'NAT-Instance'},
                    {'Key': 'Type', 'Value': 'NAT'}
                ]
            }
        ],
        # 使用Spot实例进一步节省
        InstanceMarketOptions={
            'MarketType': 'spot',
            'SpotOptions': {
                'SpotInstanceType': 'persistent',
                'InstanceInterruptionBehavior': 'stop'
            }
        }
    )

    instance_id = response['Instances'][0]['InstanceId']

    # 更新路由表
    update_route_tables_for_nat(instance_id)

    print(f"NAT Instance deployed: {instance_id}")
    print("Estimated monthly savings: $50-100")

    return instance_id

def cleanup_unused_eips():
    """AWS EC2成本优化 - 清理未使用的弹性IP"""

    ec2 = boto3.client('ec2')

    # 获取所有弹性IP
    eips = ec2.describe_addresses()

    unused_eips = []
    total_savings = 0

    for eip in eips['Addresses']:
        if 'InstanceId' not in eip and 'NetworkInterfaceId' not in eip:
            unused_eips.append(eip)
            # 每个未使用的EIP每月费用约$3.6
            total_savings += 3.6

    print(f"Found {len(unused_eips)} unused Elastic IPs")
    print(f"Monthly cost: ${total_savings:.2f}")

    if unused_eips:
        response = input("Release unused EIPs? (yes/no): ")
        if response.lower() == 'yes':
            for eip in unused_eips:
                try:
                    if 'AllocationId' in eip:
                        # VPC EIP
                        ec2.release_address(AllocationId=eip['AllocationId'])
                    else:
                        # EC2-Classic EIP
                        ec2.release_address(PublicIp=eip['PublicIp'])
                    print(f"Released: {eip.get('PublicIp', 'N/A')}")
                except Exception as e:
                    print(f"Error releasing EIP: {e}")

            print(f"nTotal monthly savings: ${total_savings:.2f}")
            print(f"Annual savings: ${total_savings * 12:.2f}")

def consolidate_load_balancers():
    """AWS EC2成本优化 - 整合负载均衡器"""

    elb = boto3.client('elbv2')

    # 获取所有ALB
    load_balancers = elb.describe_load_balancers()

    # 分析使用率
    underutilized_albs = []

    for lb in load_balancers['LoadBalancers']:
        lb_arn = lb['LoadBalancerArn']

        # 获取目标组
        target_groups = elb.describe_target_groups(
            LoadBalancerArn=lb_arn
        )

        total_targets = 0
        for tg in target_groups['TargetGroups']:
            targets = elb.describe_target_health(
                TargetGroupArn=tg['TargetGroupArn']
            )
            total_targets += len(targets['TargetHealthDescriptions'])

        # 如果目标少于3个，标记为未充分利用
        if total_targets < 3:
            underutilized_albs.append({
                'Name': lb['LoadBalancerName'],
                'Targets': total_targets,
                'MonthlyCost': 18  # ALB基础费用
            })

    if underutilized_albs:
        print("Underutilized ALBs found:")
        total_savings = 0
        for alb in underutilized_albs:
            print(f"- {alb['Name']}: {alb['Targets']} targets")
            total_savings += alb['MonthlyCost']

        print(f"nPotential monthly savings by consolidation: ${total_savings:.2f}")

        # 建议整合方案
        print("nRecommendation: Use path-based routing on a single ALB")
        print("Implementation: Create multiple target groups with path patterns")

    return underutilized_albs

class EC2CostOptimizer:
    """AWS EC2成本优化 - EC2-Other综合优化器"""

    def __init__(self):
        self.ec2 = boto3.client('ec2')
        self.cw = boto3.client('cloudwatch')
        self.ce = boto3.client('ce')

    def analyze_ec2_other_costs(self):
        """分析EC2-Other成本构成"""

        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=30)

        response = self.ce.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.isoformat(),
                'End': end_date.isoformat()
            },
            Granularity='MONTHLY',
            Metrics=['UnblendedCost'],
            GroupBy=[
                {'Type': 'DIMENSION', 'Key': 'USAGE_TYPE'}
            ],
            Filter={
                'Dimensions': {
                    'Key': 'SERVICE',
                    'Values': ['Amazon Elastic Compute Cloud - Compute']
                }
            }
        )

        ec2_other_breakdown = {}

        for result in response['ResultsByTime']:
            for group in result['Groups']:
                usage_type = group['Keys'][0]
                cost = float(group['Metrics']['UnblendedCost']['Amount'])

                if 'EBS' in usage_type:
                    category = 'EBS'
                elif 'DataTransfer' in usage_type:
                    category = 'Data Transfer'
                elif 'ElasticIP' in usage_type:
                    category = 'Elastic IP'
                elif 'NatGateway' in usage_type:
                    category = 'NAT Gateway'
                elif 'LoadBalancer' in usage_type:
                    category = 'Load Balancer'
                else:
                    category = 'Other'

                if category not in ec2_other_breakdown:
                    ec2_other_breakdown[category] = 0
                ec2_other_breakdown[category] += cost

        return ec2_other_breakdown

    def generate_optimization_report(self):
        """生成优化报告和建议"""

        costs = self.analyze_ec2_other_costs()
        recommendations = []
        total_savings = 0

        # EBS优化建议
        if costs.get('EBS', 0) > 100:
            savings = costs['EBS'] * 0.3  # 预计节省30%
            recommendations.append({
                'Category': 'EBS',
                'Current Cost': costs['EBS'],
                'Potential Savings': savings,
                'Actions': [
                    'Convert gp2 to gp3',
                    'Delete unattached volumes',
                    'Implement snapshot lifecycle',
                    'Use appropriate volume types'
                ]
            })
            total_savings += savings

        # 数据传输优化建议
        if costs.get('Data Transfer', 0) > 50:
            savings = costs['Data Transfer'] * 0.4
            recommendations.append({
                'Category': 'Data Transfer',
                'Current Cost': costs['Data Transfer'],
                'Potential Savings': savings,
                'Actions': [
                    'Create VPC endpoints',
                    'Optimize cross-AZ traffic',
                    'Use CloudFront for static content',
                    'Implement data compression'
                ]
            })
            total_savings += savings

        # 生成报告
        print("=" * 60)
        print("AWS EC2 COST OPTIMIZATION REPORT - EC2-OTHER ANALYSIS")
        print("=" * 60)
        print(f"nCurrent EC2-Other Monthly Costs:")
        for category, cost in costs.items():
            print(f"  {category}: ${cost:.2f}")

        print(f"nTotal EC2-Other: ${sum(costs.values()):.2f}")

        print("nOptimization Recommendations:")
        for rec in recommendations:
            print(f"n{rec['Category']}:")
            print(f"  Current: ${rec['Current Cost']:.2f}")
            print(f"  Potential Savings: ${rec['Potential Savings']:.2f}")
            print("  Actions:")
            for action in rec['Actions']:
                print(f"    - {action}")

        print(f"n" + "=" * 60)
        print(f"TOTAL POTENTIAL MONTHLY SAVINGS: ${total_savings:.2f}")
        print(f"ANNUAL SAVINGS: ${total_savings * 12:.2f}")
        print("=" * 60)

        return recommendations

    def execute_optimizations(self, auto_apply=False):
        """执行优化操作"""

        optimizations = [
            ('EBS Volume Type Optimization', self.optimize_ebs_volumes),
            ('Unused EIP Cleanup', self.cleanup_unused_eips),
            ('Snapshot Cleanup', self.cleanup_old_snapshots),
            ('VPC Endpoint Creation', self.create_vpc_endpoints),
            ('Load Balancer Consolidation', self.consolidate_load_balancers)
        ]

        for name, func in optimizations:
            print(f"nExecuting: {name}")
            if auto_apply:
                func()
            else:
                response = input(f"Apply {name}? (yes/no): ")
                if response.lower() == 'yes':
                    func()

# 使用示例
if __name__ == "__main__":
    optimizer = EC2CostOptimizer()

    # 生成分析报告
    optimizer.generate_optimization_report()

    # 执行优化
    optimizer.execute_optimizations(auto_apply=False)

def setup_cost_monitoring():
    """设置EC2-Other成本监控"""

    cw = boto3.client('cloudwatch')

    # 创建自定义仪表板
    dashboard_body = {
        "widgets": [
            {
                "type": "metric",
                "properties": {
                    "metrics": [
                        ["AWS/Billing", "EstimatedCharges", 
                         {"stat": "Maximum", "label": "Total Cost"}],
                        ["...", {"stat": "Maximum", "label": "EC2-Other"}]
                    ],
                    "period": 86400,
                    "stat": "Maximum",
                    "region": "us-east-1",
                    "title": "EC2-Other Cost Trend"
                }
            }
        ]
    }

    cw.put_dashboard(
        DashboardName='EC2-Other-Optimization',
        DashboardBody=json.dumps(dashboard_body)
    )

    # 设置成本异常告警
    cw.put_metric_alarm(
        AlarmName='EC2-Other-Cost-Spike',
        ComparisonOperator='GreaterThanThreshold',
        EvaluationPeriods=1,
        MetricName='EstimatedCharges',
        Namespace='AWS/Billing',
        Period=86400,
        Statistic='Maximum',
        Threshold=500.0,  # 根据实际情况调整
        ActionsEnabled=True,
        AlarmActions=['arn:aws:sns:region:account:topic'],
        AlarmDescription='Alert when EC2-Other costs spike'
    )

第1周：分析和评估
- 运行成本分析脚本
- 识别优化机会
- 计算ROI

第2周：快速优化
- 清理未使用资源
- 转换EBS卷类型
- 实施快照策略

第3-4周：架构优化
- 部署VPC端点
- 优化数据传输
- 考虑NAT实例

持续：监控和调优
- 每周成本审查
- 自动化优化脚本
- 持续改进