第12章 云成本优化¶

import boto3
from datetime import datetime, timedelta

def analyze_costs():
    """分析云成本"""
    ce = boto3.client('ce', region_name='us-east-1')

    # 查询最近30天的成本
    end_date = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')

    # 按服务查询成本
    response = ce.get_cost_and_usage(
        TimePeriod={
            'Start': start_date,
            'End': end_date
        },
        Granularity='DAILY',
        Metrics=['BlendedCost', 'UnblendedCost'],
        GroupBy=[
            {
                'Type': 'DIMENSION',
                'Key': 'SERVICE'
            }
        ]
    )

    # 分析结果
    print("成本分析报告")
    print("=" * 50)
    total_cost = 0

    for result in response['ResultsByTime']:
        date = result['TimePeriod']['Start']
        print(f"\n日期: {date}")

        for group in result['Groups']:
            service = group['Keys'][0]
            cost = float(group['Metrics']['BlendedCost']['Amount'])
            total_cost += cost
            print(f"  {service}: ${cost:.2f}")

    print(f"\n总成本: ${total_cost:.2f}")

    return total_cost

def set_budget_alert():
    """设置成本告警"""
    budgets = boto3.client('budgets', region_name='us-east-1')

    # 创建月度预算
    budget = {
        'AccountId': '123456789012',
        'Budget': {
            'BudgetName': 'MonthlyCostBudget',
            'BudgetLimit': {
                'Amount': '1000.00',
                'Unit': 'USD'
            },
            'TimeUnit': 'MONTHLY',
            'BudgetType': 'COST',
            'CostFilters': {
                'Service': ['Amazon EC2', 'Amazon RDS', 'Amazon S3']
            }
        },
        'NotificationsWithSubscribers': [
            {
                'Notification': {
                    'NotificationType': 'ACTUAL',
                    'ComparisonOperator': 'GREATER_THAN',
                    'Threshold': 80.0,
                    'ThresholdType': 'PERCENTAGE'
                },
                'Subscribers': [
                    {
                        'SubscriptionType': 'EMAIL',
                        'Address': 'devops-team@example.com'
                    }
                ]
            }
        ]
    }

    try:  # try/except捕获异常
        budgets.create_budget(**budget)
        print("成本告警设置成功")
    except Exception as e:
        print(f"设置失败: {e}")

def optimize_instances():
    """优化实例"""
    ec2 = boto3.client('ec2', region_name='us-east-1')
    ce = boto3.client('ce', region_name='us-east-1')

    # 获取所有实例
    instances = ec2.describe_instances()

    for reservation in instances['Reservations']:
        for instance in reservation['Instances']:
            instance_id = instance['InstanceId']
            instance_type = instance['InstanceType']
            state = instance['State']['Name']

            # 检查实例状态
            if state == 'running':
                # 获取实例指标
                cloudwatch = boto3.client('cloudwatch')

                # 获取CPU使用率
                cpu_metrics = cloudwatch.get_metric_statistics(
                    Namespace='AWS/EC2',
                    MetricName='CPUUtilization',
                    Dimensions=[
                        {
                            'Name': 'InstanceId',
                            'Value': instance_id
                        }
                    ],
                    StartTime=datetime.now() - timedelta(days=7),
                    EndTime=datetime.now(),
                    Period=86400,
                    Statistics=['Average']
                )

                if cpu_metrics['Datapoints']:
                    avg_cpu = cpu_metrics['Datapoints'][0]['Average']

                    # 如果CPU使用率低，建议降级
                    if avg_cpu < 20:
                        print(f"实例 {instance_id} CPU使用率低 ({avg_cpu}%)，建议降级")

class FinOpsManager:
    """FinOps管理类"""

    def __init__(self):
        self.ce = boto3.client('ce', region_name='us-east-1')
        self.budgets = boto3.client('budgets', region_name='us-east-1')

    def analyze_cost_trends(self):
        """分析成本趋势"""
        end_date = datetime.now().strftime('%Y-%m-%d')
        start_date = (datetime.now() - timedelta(days=90)).strftime('%Y-%m-%d')

        response = self.ce.get_cost_and_usage(
            TimePeriod={
                'Start': start_date,
                'End': end_date
            },
            Granularity='MONTHLY',
            Metrics=['BlendedCost']
        )

        print("成本趋势分析")
        print("=" * 50)

        for result in response['ResultsByTime']:
            period = result['TimePeriod']['Start']
            cost = float(result['Total']['BlendedCost']['Amount'])
            print(f"{period}: ${cost:.2f}")

    def implement_optimization(self):
        """实施优化策略"""
        # 1. 设置预算告警
        self.set_budget_alert()

        # 2. 优化实例
        self.optimize_instances()

        # 3. 清理未使用资源
        self.cleanup_unused_resources()

        # 4. 实施自动扩缩容
        self.implement_auto_scaling()