RJ的运维工作笔记

引言

这是一个序列操作指南文档，本篇是第一篇,玩转亚马逊云监控服务Cloudwatch之一：批量为 EC2 创建 CPU 使用率告警。在当今的云计算时代，企业为了追求更高效、更灵活的 IT 基础设施，常常会选择将业务系统迁移到公有云平台。某新能源领域的头部企业，出于对全球市场布局和技术创新的考量，决定将其业务系统从阿里云迁移到 AWS。然而，在迁移过程中，他们遇到了一个挑战：AWS CloudWatch 原生功能中没有直接支持批量添加 EC2 监控告警的功能。这对于拥有大量 EC2 实例的企业来说，手动为每台实例设置告警无疑是一项耗时且容易出错的任务。作为该企业的现场支持SA，我们通过编写一个 Python 脚本，成功地帮助他们实现了批量创建 EC2 CPU 使用量告警的功能。

背景与挑战

该新能源企业在其业务系统迁移至 AWS 后，面临着如何高效监控大量 EC2 实例性能的问题。特别是，他们需要为每一台 EC2 实例创建 CPU 使用率告警，以便在 CPU 使用率超过一定阈值时能够及时收到通知，从而快速响应潜在的性能问题。然而，AWS CloudWatch 作为 AWS 提供的强大的监控与可观测性服务，虽然功能丰富，但并没有直接提供批量创建 EC2 告警的原生功能。这意味着企业无法通过简单的点击操作来为所有 EC2 实例统一设置告警，而是需要针对每一台实例单独进行配置。对于拥有数十台甚至上百台 EC2 实例的企业来说，这种手动配置的方式不仅效率低下，而且容易因人为疏忽而导致配置不一致或遗漏某些实例，从而影响整体监控效果和系统稳定性。

解决方案设计

为了解决这一挑战，我们决定利用 AWS 提供的软件开发工具包（SDK）—— Boto3，编写一个 Python 脚本来实现批量创建 EC2 CPU 使用量告警的功能。Boto3 是 AWS 的 Python 库，允许开发者通过编写代码与 AWS 服务进行交互，从而实现自动化和定制化的操作。通过使用 Boto3，我们可以调用 AWS 的 API 来获取所有运行的 EC2 实例信息，并为每个实例创建自定义的 CPU 使用率告警。这样，企业就能够以自动化的方式一次性为所有目标实例配置统一的告警规则，不仅节省了大量时间，还确保了告警配置的一致性和准确性。

Python 脚本实现

以下是我们为企业编写的 Python 脚本，用于批量创建 EC2 CPU 使用量告警：

"""
author: RJ.Wang
Date: 2025-03-13
email: wangrenjun@gmail.com
Description: Batch creation of AWS EC2 CPU utilization alarms 
"""
import boto3
import logging
from botocore.exceptions import ClientError

# 配置日志记录
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)

AWS_REGION = "ap-southeast-1"  # 需要修改为您的 AWS 区域
sns_topic_arn = "arn:aws:sns:xxxxxx:xxxxxx:xxxxxx"  # 需要修改为您的 SNS 主题 ARN

ec2_client = boto3.client("ec2", region_name=AWS_REGION)
cw_client = boto3.client("cloudwatch", region_name=AWS_REGION)
sts_client = boto3.client("sts")

# 获取账号ID并缓存
account_id = sts_client.get_caller_identity()["Account"]

def get_instance_ids():
    """
    获取所有运行的EC2实例ID
    """
    paginator = ec2_client.get_paginator("describe_instances")
    instance_ids = []
    try:
        for page in paginator.paginate():
            for reservation in page.get("Reservations", []):
                for instance in reservation.get("Instances", []):
                    instance_id = instance.get("InstanceId")
                    state = instance.get("State", {}).get("Name", "unknown")
                    if state == "running":
                        instance_ids.append(instance_id)
                        logger.debug(f"Found running instance: {instance_id}")
                    else:
                        logger.debug(f"Skipping instance in state {state}: {instance_id}")
    except ClientError as e:
        logger.error(f"Error getting instance IDs: {e}")
    except Exception as e:
        logger.error(f"Unexpected error getting instances: {e}")
    return instance_ids

def delete_existing_alarm(alarm_name):
    """
    删除已存在的同名报警
    """
    try:
        response = cw_client.describe_alarms(AlarmNames=[alarm_name])
        if response.get("MetricAlarms"):
            cw_client.delete_alarms(AlarmNames=[alarm_name])
            logger.info(f"Deleted existing alarm: {alarm_name}")
        else:
            logger.info(f"No existing alarm found: {alarm_name}")
    except ClientError as e:
        logger.error(f"Error deleting alarm {alarm_name}: {e}")

def create_cpu_alarm(instance_id, created_alarms):
    """
    为指定实例创建CPU使用率报警
    """
    alarm_name = f"CPU-{instance_id}-Alarm"
    try:
        delete_existing_alarm(alarm_name)
        cw_client.put_metric_alarm(
            AlarmName=alarm_name,
            MetricName="CPUUtilization",
            Namespace="AWS/EC2",
            Statistic="Average",
            Period=60,
            EvaluationPeriods=2,
            Threshold=85.0,
            ComparisonOperator="GreaterThanThreshold",
            AlarmActions=[sns_topic_arn],
            OKActions=[sns_topic_arn],
            AlarmDescription=f"Alarm for CPUUtilization on instance {instance_id}",  # 增加报警描述
            Dimensions=[{"Name": "InstanceId", "Value": instance_id}]
        )
        logger.info(f"Created CPU alarm: {alarm_name}")

        alarm_arn = f"arn:aws:cloudwatch:{AWS_REGION}:{account_id}:alarm:{alarm_name}"
        created_alarms.append(alarm_arn)
    except ClientError as e:
        logger.error(f"Error creating alarm for instance {instance_id}: {e}")

def main():
    """
    主函数，执行整个流程
    """
    try:
        logger.info("Starting program...")
        instance_ids = get_instance_ids()
        if not instance_ids:
            logger.info("No running instances found. Exiting.")
            return

        logger.info(f"Found {len(instance_ids)} running instances: {instance_ids}")

        created_alarms = []
        for instance_id in instance_ids:
            create_cpu_alarm(instance_id, created_alarms)

        total_alarms = len(created_alarms)
        logger.info(f"\nTotal new alarms created: {total_alarms}")
        logger.info("Alarm ARNs:")
        for arn in created_alarms:
            logger.info(arn)

        logger.info("Program completed successfully.")
    except Exception as e:
        logger.error(f"An unexpected error occurred: {e}", exc_info=True)

if __name__ == "__main__":
    main()

配置 SNS 主题和订阅

为了确保告警通知能够及时发送到指定的电子邮件地址，我们需要配置 SNS 主题和订阅。以下是详细的步骤：

创建 SNS 主题：
- 登录 AWS 管理控制台，选择 SNS 服务。
- 在左侧导航栏中，选择主题，然后点击 创建主题。
- 输入主题名称（例如：ec2-cpu-alarms）和显示名称（可选），然后点击 创建主题。
- 记录下创建的主题 ARN，例如：arn:aws:sns:ap-southeast-1:123456789012:ec2-cpu-alarms。
订阅 SNS 主题：
- 在 SNS 控制台中，选择刚才创建的主题，点击订阅选项卡，然后点击 创建订阅。
- 在协议下拉菜单中选择 电子邮件。
- 在 终端节点 中输入要接收告警通知的电子邮件地址。
- 点击 创建订阅。
确认订阅：
- AWS 会向指定的电子邮件地址发送一封确认邮件。
- 打开邮件，点击其中的 确认订阅 链接，完成订阅过程。
- 确认后，该电子邮件地址将能够接收来自 SNS 主题的告警通知。

iShot_2025-03-13_21.25.48

脚本解析

初始化 AWS 客户端：脚本首先初始化了 EC2、CloudWatch 和 STS 客户端，用于与相应的 AWS 服务进行交互。通过 STS 客户端获取当前账号的 ID，以便后续构造报警的 ARN。
获取运行的实例 ID：get_instance_ids 函数使用 EC2 客户端的分页器获取所有运行状态的 EC2 实例 ID。它遍历每个页面的实例，筛选出状态为 “running” 的实例，并收集它们的 ID。
删除现有报警：在为每个实例创建新报警之前，delete_existing_alarm 函数会检查是否存在同名的现有报警。如果存在，则删除它，以避免冲突。
创建 CPU 报警：create_cpu_alarm 函数负责创建 CPU 使用率报警。它定义了报警的名称、指标名称、命名空间、统计方式、周期、评估周期、阈值、比较运算符、报警动作、OK 动作、描述以及维度等参数。通过调用 CloudWatch 客户端的 put_metric_alarm 方法，将报警规则发送到 AWS 服务。
主函数执行流程：main 函数作为程序的入口，协调整个流程。它首先获取运行的实例 ID 列表，然后为每个实例调用 create_cpu_alarm 函数创建报警，并收集创建的报警 ARN，最后输出总结信息。

实施步骤

配置 SNS 主题和订阅：按照上述步骤创建 SNS 主题，并通过电子邮件地址订阅该主题，确保订阅已确认。
更新脚本中的变量：在脚本中，将 AWS_REGION 变量的值替换为您的 AWS 区域，将 sns_topic_arn 变量的值替换为您创建的 SNS 主题 ARN。
在 AWS CloudShell 中测试脚本：
- 登录到 AWS 管理控制台，使用具有必要权限的 IAM 账户进行登录。确保该账户具有访问 AWS CloudShell 和使用相关 AWS 服务的权限。
- 在 AWS 管理控制台中，启动 AWS CloudShell。您可以通过点击控制台右上角的 CloudShell 图标来启动它。
- 在 CloudShell 中，克隆包含脚本的 GitHub 仓库，或者直接将脚本复制到 CloudShell 环境中。
- 运行脚本：在 CloudShell 中运行 python create_ec2_alarms.py 命令。