首页 / 新闻资讯 / 行业动态 / AWS云开户闲置资源识别:自动清理脚本与监控方案

AWS云开户闲置资源识别:自动清理脚本与监控方案

发布时间:2026.04.08

随着企业上云规模的快速扩张,AWS账户中因测试环境遗留、资源创建后忘记删除、自动扩缩容残留等原因产生的闲置资源,已成为云成本浪费的主要来源之一。据Gartner统计,企业云支出中平均有30%-45%被闲置或未充分利用的资源消耗。本文系统梳理了AWS平台上12类常见闲置资源的精准识别指标,提供了基于Boto3的可直接部署的自动化清理脚本集,设计了一套覆盖资源全生命周期的监控告警架构,并总结了分级清理、标签治理、风险隔离等最佳实践,帮助企业在保障业务连续性的前提下,实现云成本的精细化管控。

一、AWS闲置资源管理的必要性

1. 云成本浪费的严峻现状
AWS按需付费的模式虽然降低了企业的IT基础设施门槛,但也带来了"看不见的成本"问题。与传统数据中心一次性投入不同,云资源按秒计费的特性使得闲置资源会持续产生费用。一个闲置的t3.large实例每月会产生约60美元的费用,一个未挂载的EBS卷每月每GB收费0.08美元,而一个被遗忘的RDS实例每月费用可能高达数千美元。

对于拥有多个AWS账户、多个环境的中大型企业而言,分散在各个区域、各个账户中的闲置资源累积起来,每年造成的成本浪费可达数十万甚至数百万美元。更严重的是,闲置资源不仅浪费资金,还会增加安全攻击面,降低云账户的可管理性。

2. 闲置资源的主要来源
AWS账户中的闲置资源主要来源于以下几个方面:

3. 自动化管理的价值
传统的人工巡检方式效率低下,且容易遗漏。通过自动化工具实现闲置资源的识别、标记、通知和清理,可以:

二、AWS常见闲置资源类型及识别指标

精准识别是闲置资源管理的第一步。不同类型的资源有不同的闲置判断标准,需要结合其使用特性和业务场景制定合理的识别指标。

1. 计算类资源

资源类型 核心闲置指标 推荐判断阈值 注意事项
EC2 实例 CPU 利用率、网络流量、磁盘 IO 连续 7 天 CPU 平均利用率 < 5%,网络出入流量 < 100KB/s 排除批处理任务、备用实例
停止的 EC2 实例 停止时间 停止超过 30 天 排除预留的备用实例
Lambda 函数 调用次数、错误率 连续 30 天无调用记录 排除低频触发的函数
ECS/EKS 集群 服务数量、任务运行数 集群中无运行的服务和任务超过 7 天 排除预留的空集群

2. 存储类资源

资源类型 核心闲置指标 推荐判断阈值 注意事项
未挂载 EBS 卷 挂载状态、最后挂载时间 未挂载且最后挂载时间超过 30 天 排除用于备份的卷
EBS 快照 创建时间、关联卷状态 创建超过 90 天且关联卷已删除 保留合规要求的快照
S3 存储桶 对象数量、最后访问时间 连续 90 天无读写操作 排除归档存储桶
EFS 文件系统 连接数、IOPS 连续 7 天无连接且 IOPS 为 0 排除低频访问的文件系统

3.  网络类资源

资源类型 核心闲置指标 推荐判断阈值 注意事项
弹性 IP 关联状态 未关联任何实例或网络接口超过 24 小时 AWS 对未使用的弹性 IP 收取费用
负载均衡器 后端实例数、请求数 后端无健康实例且连续 7 天无请求 排除备用负载均衡器
NAT 网关 网络流量 连续 7 天出入流量 < 1KB/s 排除备用 NAT 网关
安全组 关联资源数 未关联任何实例、网络接口或负载均衡器 排除默认安全组

4. 数据库与中间件类资源

资源类型 核心闲置指标 推荐判断阈值 注意事项
RDS 实例 CPU 利用率、连接数、IOPS 连续 7 天 CPU 平均利用率 < 5% 且无连接 排除备用数据库实例
停止的 RDS 实例 停止时间 停止超过 30 天 AWS 会自动删除停止超过 7 天的 RDS 实例
ElastiCache 集群 CPU 利用率、连接数、命令数 连续 7 天无连接且命令数为 0 排除备用缓存集群
Redshift 集群 查询数、CPU 利用率 连续 7 天无查询且 CPU 利用率 < 1% 排除数据仓库备份集群

5. 其他常见闲置资源

三、自动化识别与清理脚本实现

基于Python和Boto3库,我们可以实现AWS闲置资源的自动化识别和清理。以下脚本采用"先标记、再通知、后清理"的安全策略,最大限度降低误删风险。

1. 环境准备与权限配置
首先,安装必要的依赖库:

pip install boto3 python-dotenv

创建一个具有最小权限的IAM角色,仅授予脚本所需的权限。以下是IAM策略示例:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "ec2:DescribeInstances",
                "ec2:DescribeVolumes",
                "ec2:DescribeSnapshots",
                "ec2:DescribeAddresses",
                "ec2:DescribeSecurityGroups",
                "ec2:CreateTags",
                "ec2:TerminateInstances",
                "ec2:DeleteVolume",
                "ec2:DeleteSnapshot",
                "ec2:ReleaseAddress",
                "rds:DescribeDBInstances",
                "rds:DeleteDBInstance",
                "s3:ListBuckets",
                "s3:GetBucketTagging",
                "s3:PutBucketTagging",
                "s3:DeleteBucket",
                "cloudwatch:GetMetricStatistics"
            ],
            "Resource": "*"
        }
    ]
}

2. 通用脚本框架
创建一个通用的AWS资源管理器类,封装常用的操作:

import boto3
import datetime
import logging
from typing import List, Dict, Any
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AWSResourceCleaner:
    def __init__(self, region: str = 'us-east-1', dry_run: bool = True):
        self.region = region
        self.dry_run = dry_run
        self.ec2 = boto3.client('ec2', region_name=region)
        self.rds = boto3.client('rds', region_name=region)
        self.s3 = boto3.client('s3', region_name=region)
        self.cloudwatch = boto3.client('cloudwatch', region_name=region)
        
        # 白名单:永远不清理的资源ID
        self.whitelist = set([
            # 'i-0123456789abcdef0',
            # 'vol-0123456789abcdef0'
        ])
        
        # 标记键:用于标记待清理资源
        self.tag_key = 'CleanupCandidate'
        self.tag_value = 'True'
        self.mark_date_tag = 'MarkedForCleanupDate'
    def is_whitelisted(self, resource_id: str) -> bool:
        """检查资源是否在白名单中"""
        return resource_id in self.whitelist
    def tag_resource(self, resource_id: str, resource_type: str) -> None:
        """标记资源为待清理"""
        if self.dry_run:
            logger.info(f"[DRY RUN] 标记{resource_type} {resource_id}为待清理")
            return
            
        tags = [
            {'Key': self.tag_key, 'Value': self.tag_value},
            {'Key': self.mark_date_tag, 'Value': datetime.datetime.now().isoformat()}
        ]
        
        try:
            if resource_type == 'instance':
                self.ec2.create_tags(Resources=[resource_id], Tags=tags)
            elif resource_type == 'volume':
                self.ec2.create_tags(Resources=[resource_id], Tags=tags)
            elif resource_type == 'snapshot':
                self.ec2.create_tags(Resources=[resource_id], Tags=tags)
            elif resource_type == 'eip':
                self.ec2.create_tags(Resources=[resource_id], Tags=tags)
            logger.info(f"已标记{resource_type} {resource_id}为待清理")
        except Exception as e:
            logger.error(f"标记{resource_type} {resource_id}失败: {str(e)}")
    def get_metric_average(self, namespace: str, metric_name: str, 
                          dimensions: List[Dict[str, str]], 
                          days: int = 7) -> float:
        """获取指定时间范围内的指标平均值"""
        end_time = datetime.datetime.now()
        start_time = end_time - datetime.timedelta(days=days)
        
        try:
            response = self.cloudwatch.get_metric_statistics(
                Namespace=namespace,
                MetricName=metric_name,
                Dimensions=dimensions,
                StartTime=start_time,
                EndTime=end_time,
                Period=3600,
                Statistics=['Average']
            )
            
            datapoints = response['Datapoints']
            if not datapoints:
                return 0.0
                
            return sum(d['Average'] for d in datapoints) / len(datapoints)
        except Exception as e:
            logger.error(f"获取指标失败: {str(e)}")
            return 0.0

3. 分资源类型清理脚本
基于上述通用框架,实现各类资源的识别和清理功能:

(1)闲置EC2实例识别与清理

def find_idle_ec2_instances(self, cpu_threshold: float = 5.0, days: int = 7) -> List[str]:
    """识别闲置的EC2实例"""
    idle_instances = []
    
    try:
        response = self.ec2.describe_instances(
            Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
        )
        
        for reservation in response['Reservations']:
            for instance in reservation['Instances']:
                instance_id = instance['InstanceId']
                
                if self.is_whitelisted(instance_id):
                    continue
                    
                # 检查CPU利用率
                cpu_avg = self.get_metric_average(
                    'AWS/EC2', 'CPUUtilization',
                    [{'Name': 'InstanceId', 'Value': instance_id}],
                    days
                )
                
                # 检查网络流量
                network_in_avg = self.get_metric_average(
                    'AWS/EC2', 'NetworkIn',
                    [{'Name': 'InstanceId', 'Value': instance_id}],
                    days
                )
                
                network_out_avg = self.get_metric_average(
                    'AWS/EC2', 'NetworkOut',
                    [{'Name': 'InstanceId', 'Value': instance_id}],
                    days
                )
                
                if cpu_avg < cpu_threshold and network_in_avg < 100000 and network_out_avg < 100000:
                    idle_instances.append(instance_id)
                    self.tag_resource(instance_id, 'instance')
                    
    except Exception as e:
        logger.error(f"识别闲置EC2实例失败: {str(e)}")
        
    logger.info(f"发现{len(idle_instances)}个闲置EC2实例")
    return idle_instances
def terminate_instances(self, instance_ids: List[str]) -> None:
    """终止EC2实例"""
    if not instance_ids:
        return
        
    if self.dry_run:
        logger.info(f"[DRY RUN] 将终止以下EC2实例: {instance_ids}")
        return
        
    try:
        self.ec2.terminate_instances(InstanceIds=instance_ids)
        logger.info(f"已终止以下EC2实例: {instance_ids}")
    except Exception as e:
        logger.error(f"终止EC2实例失败: {str(e)}")

(2)未挂载EBS卷识别与清理

def find_unattached_volumes(self, days: int = 30) -> List[str]:
    """识别未挂载的EBS卷"""
    unattached_volumes = []
    cutoff_date = datetime.datetime.now() - datetime.timedelta(days=days)
    
    try:
        response = self.ec2.describe_volumes(
            Filters=[{'Name': 'status', 'Values': ['available']}]
        )
        
        for volume in response['Volumes']:
            volume_id = volume['VolumeId']
            
            if self.is_whitelisted(volume_id):
                continue
                
            # 检查最后挂载时间
            last_attach_time = None
            for attachment in volume.get('Attachments', []):
                if 'AttachTime' in attachment:
                    last_attach_time = attachment['AttachTime']
                    
            if last_attach_time and last_attach_time < cutoff_date:
                unattached_volumes.append(volume_id)
                self.tag_resource(volume_id, 'volume')
                
    except Exception as e:
        logger.error(f"识别未挂载EBS卷失败: {str(e)}")
        
    logger.info(f"发现{len(unattached_volumes)}个未挂载EBS卷")
    return unattached_volumes
def delete_volumes(self, volume_ids: List[str]) -> None:
    """删除EBS卷"""
    if not volume_ids:
        return
        
    if self.dry_run:
        logger.info(f"[DRY RUN] 将删除以下EBS卷: {volume_ids}")
        return
        
    for volume_id in volume_ids:
        try:
            self.ec2.delete_volume(VolumeId=volume_id)
            logger.info(f"已删除EBS卷: {volume_id}")
        except Exception as e:
            logger.error(f"删除EBS卷{volume_id}失败: {str(e)}")

(3)未使用弹性IP识别与释放

def find_unused_eips(self) -> List[str]:
    """识别未使用的弹性IP"""
    unused_eips = []
    
    try:
        response = self.ec2.describe_addresses()
        
        for address in response['Addresses']:
            allocation_id = address['AllocationId']
            
            if self.is_whitelisted(allocation_id):
                continue
                
            # 检查是否关联了实例或网络接口
            if 'InstanceId' not in address and 'NetworkInterfaceId' not in address:
                unused_eips.append(allocation_id)
                self.tag_resource(allocation_id, 'eip')
                
    except Exception as e:
        logger.error(f"识别未使用弹性IP失败: {str(e)}")
        
    logger.info(f"发现{len(unused_eips)}个未使用弹性IP")
    return unused_eips
def release_eips(self, allocation_ids: List[str]) -> None:
    """释放弹性IP"""
    if not allocation_ids:
        return
        
    if self.dry_run:
        logger.info(f"[DRY RUN] 将释放以下弹性IP: {allocation_ids}")
        return
        
    for allocation_id in allocation_ids:
        try:
            self.ec2.release_address(AllocationId=allocation_id)
            logger.info(f"已释放弹性IP: {allocation_id}")
        except Exception as e:
            logger.error(f"释放弹性IP{allocation_id}失败: {str(e)}")

4. 主程序与调度
创建主程序,整合所有功能,并支持通过参数控制执行模式:

def main():
    import argparse
    
    parser = argparse.ArgumentParser(description='AWS闲置资源清理工具')
    parser.add_argument('--region', default='us-east-1', help='AWS区域')
    parser.add_argument('--dry-run', action='store_true', help='试运行模式,不实际删除资源')
    parser.add_argument('--mark-only', action='store_true', help='仅标记待清理资源,不执行删除')
    parser.add_argument('--clean-marked', action='store_true', help='清理已标记的资源')
    
    args = parser.parse_args()
    
    cleaner = AWSResourceCleaner(region=args.region, dry_run=args.dry_run)
    
    if args.clean_marked:
        # 清理已标记的资源
        logger.info("开始清理已标记的资源")
        # 这里需要实现根据标记查找并清理资源的逻辑
        # 为简化示例,此处省略
    else:
        # 识别并标记闲置资源
        logger.info("开始识别闲置资源")
        idle_instances = cleaner.find_idle_ec2_instances()
        unattached_volumes = cleaner.find_unattached_volumes()
        unused_eips = cleaner.find_unused_eips()
        
        if not args.mark_only:
            logger.info("开始执行清理操作")
            # 注意:在生产环境中,建议先标记,等待7天确认后再执行清理
            # cleaner.terminate_instances(idle_instances)
            # cleaner.delete_volumes(unattached_volumes)
            # cleaner.release_eips(unused_eips)
    
    logger.info("清理任务完成")
if __name__ == "__main__":
    main()

将脚本部署为AWS Lambda函数,并通过EventBridge定期触发(例如每天凌晨2点)。在生产环境中,强烈建议采用"标记7天后再清理"的策略,给业务团队足够的时间确认和恢复误标记的资源。

四、全生命周期监控告警方案

自动化清理脚本只能解决事后问题,建立一套覆盖资源全生命周期的监控告警方案,才能从源头上预防闲置资源的产生。

1. 基于AWS原生服务的监控架构
推荐使用以下AWS原生服务构建监控架构:

2. 关键监控指标与告警阈值
设置以下关键指标的告警,及时发现潜在的闲置资源:

3. 成本异常检测
使用AWS Cost Explorer的成本异常检测功能,自动识别异常的成本增长。当检测到成本异常时,立即发送通知给运维团队,排查是否有未授权的资源创建或资源滥用情况。

此外,可以自定义成本异常检测规则,例如:

4. 可视化仪表盘
使用Amazon CloudWatch Dashboards创建统一的闲置资源监控仪表盘,展示以下信息:

通过仪表盘,管理层和运维团队可以直观地了解云资源的使用情况和成本节省效果。

五、最佳实践与风险控制

闲置资源管理是一把双刃剑,过度清理可能会影响业务连续性。因此,必须建立完善的风险控制机制。

1. 分级清理策略
根据资源的重要性和影响范围,采用分级清理策略:

2. 标签管理体系
建立完善的标签管理体系,强制要求所有资源必须添加以下标签:

通过标签,可以快速定位资源负责人,实现成本分摊,并自动清理过期的资源。

3. 审批与回滚机制
建立严格的资源清理审批流程:

4. 定期审计与优化
每月进行一次全面的云资源审计,检查:

根据审计结果,不断优化闲置资源识别指标和清理策略,提高管理效率。

AWS闲置资源管理是云成本优化的重要组成部分。通过本文介绍的自动化识别清理脚本和全生命周期监控方案,企业可以显著降低云成本,提高资源利用率。

 

中新数安拥有20年网络安全服务经验,提供构涵盖防DDos/CC攻击高防IP高防DNS游戏盾Web安全加速CDN加速视频直播加速海外服务器租用SSL证书国际云开户等服务。专业技术团队全程服务支持,如您有业务需求,欢迎联系!

 


 

相关阅读:

谷歌云开户后如何提高账号稳定性

谷歌云开户多账号操作的风险说明

阿里云国际开户区域扩展动态:中东、拉美、非洲新节点部署计划

AWS云开户 + Amazon Nova全家桶零基础入门指南

腾讯云国际开户CAM权限配置指南:用户、角色、策略最佳实践

上一篇:腾讯云国际开户数据库选型指南:CDB、MongoDB、TDSQL对比 下一篇:谷歌云开户多账号操作的风险说明
联系我们,实现安全解决方案

联系我们,实现安全解决方案

留下您的联系方式,专属顾问会尽快联系您


线

返回顶部