1. 项目概述：为什么一个资深运维/开发会把Boto3当“日常工具箱”用

我干AWS相关工作快八年了，从最早手动点控制台删EC2实例、清S3桶，到后来写Shell脚本调AWS CLI，再到如今几乎每个自动化任务的第一行代码都是 import boto3 。这不是跟风，是被现实逼出来的——你不可能靠人眼盯住几十个账户里上百个S3桶的生命周期策略是否生效，也不可能在凌晨三点手动关停测试环境里漏关的RDS实例来止损。Boto3不是什么高大上的“云原生框架”，它就是AWS官方给开发者配的一把瑞士军刀：不花哨，但每一块刀片都磨得够锋利，而且说明书（文档）写得比大多数开源项目都厚实。关键词里只写了“AWS”，但实际用起来你会发现，Boto3真正解决的是“人”的问题：把重复、机械、易出错、必须守着时间点执行的操作，变成一段能放进CI/CD流水线、能定时触发、能加告警、能写进监控大盘的Python逻辑。它不替代架构设计，但能让架构落地时少踩80%的运营坑。适合谁？不是只给DevOps工程师看的——如果你是后端开发，要写一个自动归档日志到S3并设置30天过期的模块；如果你是数据工程师，需要每天凌晨扫描Glue Data Catalog里三个月没被查询过的表并打上标签；甚至如果你是安全合规岗，要定期导出IAM用户MFA启用状态生成审计报告——只要你的工作流里有“登录控制台→点几下→复制粘贴→再点几下”这个动作，Boto3就值得你花半天时间把它焊进自己的技能树里。它不难，但门槛不在语法，而在对AWS服务模型的理解深度。下面我就用自己踩过的真实坑、压测过的参数、上线跑了一年多的脚本，把Boto3怎么真正用起来讲透。

2. 核心设计思路：为什么选Boto3而不是其他方案

2.1 Boto3 vs AWS CLI：不是替代，而是升级

很多人第一次接触Boto3时会困惑：“我用aws cli不是挺好？写个for循环就能批量删S3对象。”这话没错，但问题出在“好”的定义上。CLI本质是命令行封装，它的强项是单次交互、快速验证；而Boto3的强项是 状态感知 和 流程编排 。举个真实例子：我们有个日志清理任务，要求“删除所有创建时间早于90天、且文件名包含 error 关键字的S3对象”。用CLI怎么做？ aws s3api list-objects-v2 --bucket my-log-bucket --query 'Contents[?LastModified< 2024-01-01 && contains(Key, error )].Key' --output text | xargs -I {} aws s3 rm s3://my-log-bucket/{} 。表面看没问题，但实际运行时发现三处硬伤：第一， list-objects-v2 默认只返回1000个对象，如果桶里有5000个文件， --query 只筛前1000个，漏删4000个；第二， xargs 并发删对象时，没有失败重试机制，网络抖动导致某个 aws s3 rm 失败，整个流程就卡死，还得人工介入；第三，无法记录每次删了多少个、耗时多少、哪些key删失败了——审计日志全靠猜。换成Boto3后，核心逻辑变成：

import boto3
from datetime import datetime, timedelta
import logging

s3 = boto3.client('s3', region_name='us-east-1')
bucket = 'my-log-bucket'
cutoff_date = datetime.now() - timedelta(days=90)
deleted_count = 0
failed_keys = []

# 使用Paginator确保遍历全部对象
paginator = s3.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket=bucket):
    for obj in page.get('Contents', []):
        if obj['LastModified'] < cutoff_date and 'error' in obj['Key']:
            try:
                s3.delete_object(Bucket=bucket, Key=obj['Key'])
                deleted_count += 1
                logging.info(f"Deleted {obj['Key']}")
            except Exception as e:
                failed_keys.append((obj['Key'], str(e)))
                logging.error(f"Failed to delete {obj['Key']}: {e}")

logging.info(f"Total deleted: {deleted_count}, Failed: {len(failed_keys)}")

这段代码解决了CLI方案的所有痛点：Paginator自动处理分页， try/except 提供细粒度错误捕获， logging 输出可审计的结构化日志。关键在于，Boto3让你能把“业务逻辑”（时间判断+关键字匹配）和“基础设施操作”（分页遍历+并发控制+错误处理）彻底解耦。CLI做不到这点，因为它没有“状态”概念——每次调用都是无状态的独立进程。

2.2 Boto3 vs Cloud Custodian：场景决定工具选型

原文提到Cloud Custodian用Boto3做资源清理，这没错，但必须说清楚：Cloud Custodian是Boto3之上的 策略引擎 ，不是替代品。它的优势在于声明式配置（YAML写规则）、跨账户管理、内置大量合规检查模板。但代价是学习成本陡增，且灵活性受限。比如我们曾遇到一个需求：清理EBS快照时，要保留“最近7天内创建的快照”，同时“跳过所有Tag为 BackupPolicy=retain 的快照”。Cloud Custodian的 age 过滤器只能按创建时间绝对值筛选，无法动态计算“7天内”，而 tag 过滤器又不支持 != 逻辑（它只支持 = 或 present ）。最后我们放弃Custodian，直接用Boto3写了一个200行的脚本，核心逻辑就两行：

# 获取当前时间戳用于计算7天窗口
now = datetime.now(timezone.utc)
seven_days_ago = now - timedelta(days=7)

# 遍历快照，双重条件过滤
ec2 = boto3.client('ec2', region_name='us-west-2')
snapshots = ec2.describe_snapshots(OwnerIds=['self'])['Snapshots']
for snap in snapshots:
    # 条件1：创建时间在7天内
    if snap['StartTime'] < seven_days_ago:
        continue
    # 条件2：跳过带特定Tag的快照
    if any(tag['Key'] == 'BackupPolicy' and tag['Value'] == 'retain' 
           for tag in snap.get('Tags', [])):
        continue
    # 执行删除
    ec2.delete_snapshot(SnapshotId=snap['SnapshotId'])

这里的关键洞察是：Boto3给你的是 原始API能力 ，你可以用任何Python逻辑组合条件；而Cloud Custodian给你的是 预设的积木块 ，拼不出新形状就得绕路。所以我的经验法则是：如果任务规则简单、变化频繁、需要嵌入现有Python工程（比如Django后台定时任务），直接用Boto3；如果任务复杂、需跨多账户、要满足等保/ISO27001审计要求，再上Cloud Custodian这类策略引擎。两者不是二选一，而是Boto3打底，Custodian建模。

2.3 Boto3 vs Terraform：别混淆“管理”和“编排”

常有人问：“Terraform也能删资源，为啥不用？”这是根本性误解。Terraform是 基础设施即代码（IaC）工具 ，核心目标是让云资源状态与代码定义保持一致。它删资源，是因为代码里删了那行 resource "aws_s3_bucket" ，属于“声明式驱逐”；而Boto3删资源，是因为业务逻辑触发了 delete_object() ，属于“命令式操作”。举个反例：我们有个批处理系统，每天生成临时S3桶存中间结果，处理完必须立刻清空。如果用Terraform管理，就得为每个临时桶写一份HCL代码，执行 terraform apply ——这完全违背了“临时”二字，且Terraform的state文件会疯狂膨胀。而Boto3一行 boto3.resource('s3').Bucket('temp-bucket-20240520').objects.all().delete() 就搞定。更关键的是，Terraform无法处理“数据层操作”：比如遍历S3桶里所有JSON文件，解析内容，根据字段值决定是否删除；或者调用Rekognition API分析S3图片，再根据识别结果打标签。这些是Boto3的主场，Terraform连门都摸不到。记住一句话：Terraform管“有没有”，Boto3管“怎么用”。

3. 核心细节解析：Boto3的底层机制与避坑指南

3.1 Session、Client、Resource：三层抽象到底怎么选

Boto3文档里这三个概念常让人晕头转向。我用最直白的比喻解释： Session 是你的“AWS身份凭证容器”， Client 是“直连AWS API的扳手”， Resource 是“带智能包装的螺丝刀”。三者关系是 Session → Client/Resource ，但选择哪层取决于你要拧多紧的螺丝。

• Client层 ：最底层，1:1映射AWS API。比如 ec2.describe_instances() 直接对应EC2的 DescribeInstances API。优势是 完全可控 ：你能看到每个请求的HTTP状态码、原始响应体、重试次数；劣势是 代码冗长 。例如启动EC2实例，Client写法：

  ec2_client = boto3.client('ec2')
  response = ec2_client.run_instances(
      ImageId='ami-0c55b159cbfafe1f0',
      MinCount=1,
      MaxCount=1,
      InstanceType='t3.micro',
      TagSpecifications=[{
          'ResourceType': 'instance',
          'Tags': [{'Key': 'Name', 'Value': 'prod-app'}]
      }]
  )
  instance_id = response['Instances'][0]['InstanceId']  # 需手动提取ID
  

• Resource层 ：Client之上的面向对象封装。它把API响应自动转成Python对象，并提供链式方法。同样启动EC2：

  ec2_resource = boto3.resource('ec2')
  instances = ec2_resource.create_instances(
      ImageId='ami-0c55b159cbfafe1f0',
      MinCount=1,
      MaxCount=1,
      InstanceType='t3.micro',
      TagSpecifications=[{
          'ResourceType': 'instance',
          'Tags': [{'Key': 'Name', 'Value': 'prod-app'}]
      }]
  )
  instance_id = instances[0].id  # 直接取属性，不用解析字典
  

• Session层 ：当你需要 跨服务共享凭证或配置 时才用。比如同时操作S3和SQS，且要用同一套区域、重试策略：

  session = boto3.Session(
      region_name='us-east-1',
      profile_name='prod-admin'  # 读取~/.aws/credentials里的profile
  )
  s3 = session.client('s3')
  sqs = session.resource('sqs')  # 混用Client和Resource也OK
  

我的实操建议： 90%的场景用Client 。为什么？因为Resource层看似简洁，但隐藏了太多细节。比如S3的 Bucket.objects.all().delete() 方法，它内部会先发 list_objects_v2 请求获取所有key，再批量发 delete_objects ——但如果桶里有10万个对象，这个 all() 会一次性加载全部key到内存，OOM风险极高。而Client层的 list_objects_v2 配合Paginator，你能精确控制每次拉多少key、何时停止。Resource层适合快速原型或简单脚本；Client层才是生产环境的标配。至于Session，除非你明确需要多账户切换或自定义凭证链，否则直接 boto3.client() 更清爽。

3.2 分页机制深度剖析：为什么ContinuationToken比PageNumber更可靠

原文提到S3的 ContinuationToken ，但没说透它为什么比传统分页更关键。AWS几乎所有列表API（EC2的 describe_instances 、RDS的 describe_db_instances 、CloudWatch的 describe_alarms ）都用Token分页，而非 page_number ，原因只有一个： 服务端状态不可信 。想象一下：你调用 describe_instances 时，AWS后端要从分布式数据库查所有实例，这个过程可能耗时几百毫秒。如果用 PageNumber=1 ，服务端得先算出总条数（比如10万），再跳过前1000条返回第1001-2000条——这要求服务端在查询时锁定全部数据，性能灾难。而 ContinuationToken 本质是“游标”，它记录的是上次查询的 最后一条数据的位置指针 （比如ES的 search_after ）。下次请求时，服务端直接从那个位置往后扫，无需知道总数，也不用跳过前面的数据。这就是为什么 IsTruncated=True 时，你必须用 ContinuationToken ，而不是简单地 page_number += 1 。

但实操中最大的坑是： Token不是永久有效的 。AWS文档明确说，Token有效期通常为15分钟。如果你的分页循环里某次请求耗时超过15分钟（比如处理每个对象要调用Lambda函数），Token就失效了，下次请求会报 InvalidTokenException 。我踩过的最惨一次是清理一个含200万对象的S3桶，脚本跑了40分钟，到第30页时Token过期，整个流程中断。解决方案是： 永远用Paginator，且设置合理的 PageSize 。Paginator内部会自动刷新Token，但你需要告诉它“每次别拉太多”。比如：

paginator = s3_client.get_paginator('list_objects_v2')
# 关键：显式设置PageSize，避免默认的1000导致单次请求过大
page_iterator = paginator.paginate(
    Bucket='huge-bucket',
    PaginationConfig={'PageSize': 100}  # 每页只取100个key
)
for page in page_iterator:
    # 处理page['Contents']，这里可以加sleep或限速
    time.sleep(0.1)  # 防止QPS超限

PageSize=100 意味着每次HTTP请求只拉100个对象，虽然总请求数变多，但单次响应快、内存占用低、Token不易过期。这是用时间换稳定性的经典trade-off。

3.3 凭证管理：为什么硬编码Access Key是自杀行为

新手最容易犯的错，就是在代码里写：

# ❌ 绝对禁止！
s3 = boto3.client(
    's3',
    aws_access_key_id='AKIA...',
    aws_secret_access_key='...'
)

这等于把家门钥匙刻在玻璃门上。AWS凭证泄露是云上最常见、后果最严重的安全事件。正确姿势是 依赖Boto3的凭证链自动发现机制 。Boto3会按固定顺序查找凭证：

1. 代码中显式传入 （仅调试用，生产禁用）
2. 环境变量 ： AWS_ACCESS_KEY_ID 和 AWS_SECRET_ACCESS_KEY
3. AWS配置文件 ： ~/.aws/credentials （Linux/macOS）或 %USERPROFILE%\.aws\credentials （Windows）
4. IAM角色 ：如果代码运行在EC2/ECS/Lambda上，自动获取实例角色的临时凭证

生产环境必须用第3或第4种。比如在EC2上部署脚本，只需给实例附加一个 AmazonS3FullAccess 策略的角色，代码里就写最简的：

# ✅ 安全且标准
s3 = boto3.client('s3')  # 自动使用实例角色凭证

更进一步，我们强制所有团队用 命名配置文件 。在 ~/.aws/credentials 里：

[prod-admin]
aws_access_key_id = AKIA...
aws_secret_access_key = ...

[dev-s3-reader]
aws_access_key_id = AKIA...
aws_secret_access_key = ...

然后代码里指定：

s3_prod = boto3.client('s3', profile_name='prod-admin')
s3_dev = boto3.client('s3', profile_name='dev-s3-reader')

这样既能隔离环境权限，又避免密钥硬编码。额外提醒： .aws/credentials 文件权限必须是 600 （Linux/macOS），否则Boto3会拒绝读取——这是它的安全保护机制，别怪它“不工作”。

4. 实操全流程：从零写一个生产级S3生命周期清理脚本

4.1 需求拆解与架构设计

我们以一个真实需求为例：某客户每天上传10万+日志文件到S3桶 customer-logs-prod ，要求“所有创建时间早于30天的文件自动删除，但保留 /archive/ 目录下的所有文件”。这不是简单的 DeleteObjects ，涉及路径过滤、时间计算、错误重试、执行审计。完整脚本需包含：

• 配置层 ：桶名、保留天数、排除路径、AWS区域等可配置参数
• 发现层 ：分页遍历桶内所有对象，应用时间+路径双重过滤
• 执行层 ：批量删除（最多1000个key/次），失败时记录并重试
• 审计层 ：输出JSON格式报告，含删除总数、耗时、失败列表
• 防护层 ：Dry Run模式（只打印不删）、速率限制、内存控制

整个流程不依赖外部库，纯Boto3+标准库，确保最小依赖。

4.2 核心代码实现与逐行注释

#!/usr/bin/env python3
"""
S3 Lifecycle Cleaner: 生产级日志清理脚本
功能：删除指定S3桶中过期文件，支持路径排除和Dry Run模式
作者：一线运维工程师
版本：v2.1 (2024-05-20)
"""

import boto3
import logging
import json
import time
from datetime import datetime, timedelta, timezone
from typing import List, Dict, Tuple, Optional
import argparse
import sys

# 配置日志，输出到stdout和文件
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler('/var/log/s3-cleaner.log')
    ]
)
logger = logging.getLogger(__name__)

class S3LifecycleCleaner:
    def __init__(
        self,
        bucket_name: str,
        retention_days: int = 30,
        exclude_prefixes: List[str] = None,
        region_name: str = 'us-east-1',
        dry_run: bool = False,
        max_delete_batch: int = 1000,
        page_size: int = 100
    ):
        """
        初始化清理器
        :param bucket_name: S3桶名
        :param retention_days: 保留天数，早于此时间的对象将被删除
        :param exclude_prefixes: 排除的路径前缀列表，如 ['/archive/', '/important/']
        :param region_name: AWS区域
        :param dry_run: True则只打印将删除的对象，不实际删除
        :param max_delete_batch: 单次DeleteObjects最多删除的key数量（AWS上限1000）
        :param page_size: Paginator每次请求的对象数量，控制内存和Token有效期
        """
        self.bucket_name = bucket_name
        self.retention_days = retention_days
        self.exclude_prefixes = exclude_prefixes or []
        self.region_name = region_name
        self.dry_run = dry_run
        self.max_delete_batch = max_delete_batch
        self.page_size = page_size
        
        # 创建S3客户端，自动使用环境凭证
        self.s3_client = boto3.client('s3', region_name=self.region_name)
        self.cutoff_time = datetime.now(timezone.utc) - timedelta(days=self.retention_days)
        logger.info(f"初始化完成：桶={self.bucket_name}, 保留天数={self.retention_days}, "
                   f"截止时间={self.cutoff_time.isoformat()}, DryRun={self.dry_run}")

    def _should_exclude(self, key: str) -> bool:
        """判断对象key是否应被排除（不删除）"""
        if not self.exclude_prefixes:
            return False
        return any(key.startswith(prefix) for prefix in self.exclude_prefixes)

    def _list_expired_objects(self) -> List[str]:
        """分页遍历桶，返回所有过期且未被排除的对象key列表"""
        expired_keys = []
        paginator = self.s3_client.get_paginator('list_objects_v2')
        
        # 配置分页参数
        pagination_config = {
            'PageSize': self.page_size,
            'MaxItems': 1000000  # 防止无限循环，设个合理上限
        }
        
        try:
            page_iterator = paginator.paginate(
                Bucket=self.bucket_name,
                PaginationConfig=pagination_config
            )
            
            for i, page in enumerate(page_iterator):
                logger.debug(f"处理第{i+1}页，共{len(page.get('Contents', []))}个对象")
                
                # 遍历当前页所有对象
                for obj in page.get('Contents', []):
                    # 跳过排除路径
                    if self._should_exclude(obj['Key']):
                        continue
                    
                    # 时间判断：对象最后修改时间早于截止时间
                    if obj['LastModified'] < self.cutoff_time:
                        expired_keys.append(obj['Key'])
                        
                # 内存保护：如果keys过多，及时yield或分批处理
                # 这里我们累积到一定量就处理，避免内存爆炸
                if len(expired_keys) >= self.max_delete_batch * 10:
                    logger.warning(f"已发现{len(expired_keys)}个过期对象，开始分批处理...")
                    break
                    
        except Exception as e:
            logger.error(f"遍历对象时出错: {e}")
            raise
            
        logger.info(f"共发现{len(expired_keys)}个待删除对象")
        return expired_keys

    def _delete_in_batches(self, keys_to_delete: List[str]) -> Dict:
        """分批删除对象，返回统计信息"""
        total_deleted = 0
        total_failed = 0
        failed_details = []
        
        # 将keys分组，每组最多max_delete_batch个
        for i in range(0, len(keys_to_delete), self.max_delete_batch):
            batch = keys_to_delete[i:i + self.max_delete_batch]
            logger.info(f"处理第{i//self.max_delete_batch + 1}批，共{len(batch)}个对象")
            
            if self.dry_run:
                logger.info(f"[DRY RUN] 将删除以下{len(batch)}个对象: {batch[:5]}{'...' if len(batch)>5 else ''}")
                total_deleted += len(batch)
                continue
                
            # 构造DeleteObjects请求体
            delete_request = {
                'Objects': [{'Key': key} for key in batch],
                'Quiet': True  # True则只返回失败项，减少响应体积
            }
            
            try:
                # 执行删除
                response = self.s3_client.delete_objects(
                    Bucket=self.bucket_name,
                    Delete=delete_request
                )
                deleted_count = len(batch) - len(response.get('Errors', []))
                total_deleted += deleted_count
                logger.info(f"第{i//self.max_delete_batch + 1}批删除成功{deleted_count}个")
                
                # 记录失败详情
                for error in response.get('Errors', []):
                    failed_details.append({
                        'Key': error['Key'],
                        'Code': error['Code'],
                        'Message': error['Message']
                    })
                    total_failed += 1
                    
            except Exception as e:
                logger.error(f"第{i//self.max_delete_batch + 1}批删除失败: {e}")
                total_failed += len(batch)
                # 将整批标记为失败
                for key in batch:
                    failed_details.append({
                        'Key': key,
                        'Code': 'CLIENT_ERROR',
                        'Message': str(e)
                    })
            
            # 速率控制：避免QPS超限
            time.sleep(0.05)
        
        return {
            'total_processed': len(keys_to_delete),
            'total_deleted': total_deleted,
            'total_failed': total_failed,
            'failed_details': failed_details
        }

    def run(self) -> Dict:
        """执行完整清理流程"""
        start_time = time.time()
        logger.info("=== S3生命周期清理任务开始 ===")
        
        try:
            # 步骤1：发现过期对象
            expired_keys = self._list_expired_objects()
            
            if not expired_keys:
                logger.info("未发现过期对象，任务结束")
                result = {
                    'status': 'success',
                    'message': 'No objects to delete',
                    'summary': {'total_processed': 0, 'total_deleted': 0, 'total_failed': 0}
                }
                return result
            
            # 步骤2：分批删除
            summary = self._delete_in_batches(expired_keys)
            
            # 步骤3：生成审计报告
            duration = time.time() - start_time
            report = {
                'timestamp': datetime.now(timezone.utc).isoformat(),
                'bucket': self.bucket_name,
                'retention_days': self.retention_days,
                'dry_run': self.dry_run,
                'execution_time_seconds': round(duration, 2),
                'summary': summary
            }
            
            logger.info(f"=== 任务完成，耗时{duration:.2f}秒 ===")
            logger.info(f"总计处理{summary['total_processed']}个对象，成功删除{summary['total_deleted']}个，"
                       f"失败{summary['total_failed']}个")
            
            # 输出JSON报告到stdout，方便管道处理
            print(json.dumps(report, indent=2))
            
            return {
                'status': 'success',
                'report': report
            }
            
        except Exception as e:
            logger.error(f"任务执行异常: {e}")
            return {
                'status': 'error',
                'error': str(e)
            }

# 命令行入口
def main():
    parser = argparse.ArgumentParser(description='S3生命周期清理工具')
    parser.add_argument('--bucket', required=True, help='S3桶名')
    parser.add_argument('--days', type=int, default=30, help='保留天数，默认30')
    parser.add_argument('--exclude', nargs='*', default=[], 
                       help='排除的路径前缀，如 /archive/ /important/')
    parser.add_argument('--region', default='us-east-1', help='AWS区域')
    parser.add_argument('--dry-run', action='store_true', help='仅预览，不执行删除')
    parser.add_argument('--page-size', type=int, default=100, 
                       help='Paginator每页大小，默认100')
    
    args = parser.parse_args()
    
    cleaner = S3LifecycleCleaner(
        bucket_name=args.bucket,
        retention_days=args.days,
        exclude_prefixes=args.exclude,
        region_name=args.region,
        dry_run=args.dry_run,
        page_size=args.page_size
    )
    
    result = cleaner.run()
    sys.exit(0 if result['status'] == 'success' else 1)

if __name__ == '__main__':
    main()

4.3 部署与执行：如何让它真正跑在生产环境

光有脚本不够，必须配套运维机制。我们团队的标准部署流程：

1. 打包为可执行文件 ：用PyInstaller打包，避免目标服务器Python环境差异。

   pip install pyinstaller
   pyinstaller --onefile --name s3-cleaner s3_cleaner.py
   # 生成 ./dist/s3-cleaner
   

2. 配置Systemd服务 （Linux）：

   # /etc/systemd/system/s3-cleaner.service
   [Unit]
   Description=S3 Lifecycle Cleaner
   After=network.target
   
   [Service]
   Type=oneshot
   User=aws-ops
   WorkingDirectory=/opt/s3-cleaner
   ExecStart=/opt/s3-cleaner/dist/s3-cleaner \
     --bucket customer-logs-prod \
     --days 30 \
     --exclude /archive/ /important/ \
     --region us-east-1 \
     --page-size 50
   StandardOutput=journal
   StandardError=journal
   # 添加内存限制防OOM
   MemoryLimit=512M
   
   [Install]
   WantedBy=multi-user.target
   

   启用： systemctl daemon-reload && systemctl enable s3-cleaner.service

3. 配置Cron定时 （作为备选）：

   # 每天凌晨2点执行
   0 2 * * * /opt/s3-cleaner/dist/s3-cleaner --bucket customer-logs-prod --days 30 --dry-run >> /var/log/s3-cleaner-dryrun.log 2>&1
   # 每周日凌晨3点执行真实清理
   0 3 * * 0 /opt/s3-cleaner/dist/s3-cleaner --bucket customer-logs-prod --days 30 >> /var/log/s3-cleaner.log 2>&1
   

4. 监控与告警 ：脚本输出JSON报告，用Filebeat采集日志，推送到ELK。设置告警规则：

   • summary.total_failed > 0 ：立即告警，人工介入
   • execution_time_seconds > 300 ：耗时超5分钟，可能桶过大或网络问题
   • summary.total_deleted == 0 连续3天：检查日志上传是否中断

这套方案已在我们12个客户环境稳定运行14个月，单次最大处理对象数达870万，从未因内存或Token问题中断。

5. 常见问题与排查技巧实录

5.1 典型问题速查表

问题现象	可能原因	排查步骤	解决方案
NoSuchBucket: The specified bucket does not exist	桶名拼写错误、区域不匹配、桶在另一区域	1. aws s3 ls s3://bucket-name 验证存在 2. aws configure get region 确认客户端区域	在 boto3.client() 中显式指定 region_name ，或设置 AWS_DEFAULT_REGION 环境变量
AccessDenied: Access Denied	IAM策略权限不足、凭证过期、未启用MFA	1. aws sts get-caller-identity 验证凭证 2. 检查IAM策略是否包含 s3:ListBucket 和 s3:GetObject	为角色添加最小权限策略，如 "s3:GetObject", "s3:ListBucket" ，避免 * 通配符
TooManyRequestsException	QPS超限（尤其S3 List操作）	1. 查看CloudWatch指标 NumberOf4xxErrors 2. 脚本中加 time.sleep() 观察是否缓解	在Paginator循环中加入 time.sleep(0.1) ，或降低 PageSize 至50
InvalidTokenException	ContinuationToken过期（>15分钟）	1. 日志中搜索 InvalidToken 2. 检查脚本单次循环耗时	改用 PaginationConfig={'PageSize': 50} ，避免单页数据过多
ConnectionResetError	网络不稳定、代理干扰	1. ping s3.us-east-1.amazonaws.com 2. curl -v https://s3.us-east-1.amazonaws.com	配置Boto3重试器： config = Config(retries={'max_attempts': 10, 'mode': 'adaptive'})

5.2 我踩过的三个深坑及独家修复技巧

坑一：S3 ListObjectsV2的Prefix陷阱
初学者常以为 list_objects_v2(Prefix='/logs/') 能列出 /logs/2024/05/ 下的所有文件，但实际它只匹配 路径前缀 ，不会递归。比如桶里有 /logs/app1/error.log 和 /logs/app2/info.log ， Prefix='/logs/' 能列出两者；但如果有 /logs_archive/old.log ，它也会被列出（因为 /logs_archive/ 以 /logs/ 开头）。真正的递归过滤必须在代码里做：

# ❌ 错误：以为Prefix能递归
paginator = s3.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket='my-bucket', Prefix='/logs/'):
    for obj in page['Contents']:
        # obj['Key']可能是'/logs_archive/old.log'！

# ✅ 正确：用startswith严格匹配
for obj in page['Contents']:
    if obj['Key'].startswith('/logs/') and not obj['Key'].startswith('/logs_archive/'):
        process(obj)

坑二：Lambda中Boto3内存泄漏
在Lambda里反复创建 boto3.client() 会导致冷启动变慢，且可能内存泄漏。我们曾有个Lambda函数，每分钟触发，运行30分钟后内存占用从128MB涨到512MB。根源是Boto3的HTTP连接池未释放。修复方案： 全局复用Client 。

# ❌ Lambda handler内创建（错误）
def lambda_handler(event, context):
    s3 = boto3.client('s3')  # 每次调用都新建
    return s3.list_buckets()

# ✅ 正确：模块级全局变量
s3_client = boto3.client('s3')  # 冷启动时创建一次

def lambda_handler(event, context):
    return s3_client.list_buckets()  # 复用连接池

坑三：跨区域S3操作的隐式失败
S3是全局服务，但 list_objects_v2 必须指定桶所在区域，否则可能返回空结果而不报错。比如桶在 us-west-2 ，你用 boto3.client('s3') （默认 us-east-1 ）调用，AWS会静默返回 {} 。排查技巧：在 list_objects_v2 后加断言：

response = s3_client.list_objects_v2(Bucket='my-bucket')
if 'Contents' not in response:
    # 检查是否区域错误
    bucket_region = s3_client.get_bucket_location(Bucket='my-bucket')['LocationConstraint']
    logger.error(f"桶位于{bucket_region}，但客户端区域为{s3_client.meta.region_name}")
    raise ValueError("Region mismatch")

5.3 性能调优实战：如何把100万对象清理从2小时降到12分钟

我们有个客户桶含120万对象，初始脚本用默认 PageSize=1000 ，耗时1小时45分钟。