如何使用boto3查询180天内未活跃S3 Bucket的最后访问时间

要实现这个需求，首先要明确：S3本身不会自动记录Bucket级别的最后访问时间，需要借助CloudWatch Metrics或S3访问日志来获取相关数据。以下是两种可行的实现方案：

方案一：通过CloudWatch Metrics判断活跃度并获取最后访问时间 #

此方案无需额外配置（S3默认会向CloudWatch发送基础访问指标），适合快速批量检查所有Bucket。

代码实现 #

import boto3
from datetime import datetime, timedelta

# 初始化Boto3客户端
s3_client = boto3.client('s3')
cloudwatch_client = boto3.client('cloudwatch')

# 定义时间范围：最近180天
end_time = datetime.utcnow()
inactive_threshold = timedelta(days=180)
start_time = end_time - inactive_threshold

# 获取所有S3 Bucket
buckets = s3_client.list_buckets()['Buckets']

for bucket in buckets:
    bucket_name = bucket['Name']
    bucket_creation_date = bucket['CreationDate']
    
    # 查询最近180天的总请求数
    metric_response = cloudwatch_client.get_metric_statistics(
        Namespace='AWS/S3',
        MetricName='AllRequests',
        Dimensions=[
            {'Name': 'BucketName', 'Value': bucket_name},
            {'Name': 'StorageType', 'Value': 'AllStorageTypes'}
        ],
        StartTime=start_time,
        EndTime=end_time,
        Period=86400,  # 按天统计
        Statistics=['Sum'],
        Unit='Count'
    )
    
    datapoints = metric_response['Datapoints']
    # 判断最近180天是否有访问活动
    has_recent_activity = any(dp['Sum'] > 0 for dp in datapoints)
    
    if not has_recent_activity:
        # 查找最后一次有活动的时间（若存在）
        last_access_time = bucket_creation_date
        
        # 扩大查询范围至Bucket创建以来的所有数据
        extended_metric_response = cloudwatch_client.get_metric_statistics(
            Namespace='AWS/S3',
            MetricName='AllRequests',
            Dimensions=[
                {'Name': 'BucketName', 'Value': bucket_name},
                {'Name': 'StorageType', 'Value': 'AllStorageTypes'}
            ],
            StartTime=bucket_creation_date,
            EndTime=start_time,
            Period=86400,
            Statistics=['Sum'],
            Unit='Count'
        )
        
        extended_datapoints = extended_metric_response['Datapoints']
        if extended_datapoints:
            # 按时间倒序排序，取最后一次有请求的时间
            sorted_dps = sorted(extended_datapoints, key=lambda x: x['Timestamp'], reverse=True)
            last_access_time = sorted_dps[0]['Timestamp']
        
        print(f"Bucket: {bucket_name}")
        print(f"最近180天状态：未活跃")
        print(f"最后访问/创建时间：{last_access_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n")

注意事项 #

• 权限要求：IAM用户需拥有s3:ListAllMyBuckets和cloudwatch:GetMetricStatistics权限。
• 指标延迟：CloudWatch指标存在约15分钟的延迟，无法实时获取最新访问记录。
• 无访问记录的Bucket：若Bucket从未有过任何请求，最后访问时间将显示为Bucket创建时间。

方案二：通过S3访问日志解析最后访问时间 #

此方案需要提前为目标Bucket配置访问日志（将日志存储到指定的日志Bucket），能提供更精确的访问记录。

代码实现 #

import boto3
from datetime import datetime, timedelta

# 初始化Boto3客户端
s3_client = boto3.client('s3')

# 配置日志Bucket信息
log_bucket_name = "your-log-bucket-name"  # 存储访问日志的Bucket
log_prefix = "s3-access-logs/"  # 日志文件的前缀路径

# 定义时间范围：最近180天
end_time = datetime.utcnow()
inactive_threshold = timedelta(days=180)
start_time = end_time - inactive_threshold

# 获取所有目标Bucket
target_buckets = s3_client.list_buckets()['Buckets']

for bucket in target_buckets:
    bucket_name = bucket['Name']
    # 构造日志文件的起始筛选前缀（匹配目标Bucket的日志）
    log_start_after = f"{log_prefix}{bucket_name}-{start_time.strftime('%Y-%m-%d')}"
    
    # 列出最近180天内的日志文件
    log_response = s3_client.list_objects_v2(
        Bucket=log_bucket_name,
        Prefix=f"{log_prefix}{bucket_name}-",
        StartAfter=log_start_after
    )
    
    log_files = log_response.get('Contents', [])
    if not log_files:
        print(f"Bucket: {bucket_name}")
        print(f"最近180天状态：未活跃（无访问日志记录）\n")
        continue
    
    # 取最新的日志文件
    latest_log_file = sorted(log_files, key=lambda x: x['LastModified'], reverse=True)[0]
    
    # 下载并解析日志文件
    log_content = s3_client.get_object(
        Bucket=log_bucket_name,
        Key=latest_log_file['Key']
    )['Body'].read().decode('utf-8')
    
    # 解析日志中的最后访问时间（日志格式需与实际配置匹配）
    # 示例日志格式：192.168.1.1 - - [05/Mar/2024:14:30:00 +0000] "GET /bucket/obj.txt HTTP/1.1" 200 ...
    last_access_line = log_content.strip().split('\n')[-1]
    time_str = last_access_line.split('[')[1].split(']')[0]
    last_access_time = datetime.strptime(time_str, '%d/%b/%Y:%H:%M:%S %z')
    
    print(f"Bucket: {bucket_name}")
    print(f"最后访问时间：{last_access_time.strftime('%Y-%m-%d %H:%M:%S %Z')}\n")

注意事项 #

• 日志配置：需提前为目标Bucket开启访问日志，指定日志存储的Bucket和前缀。
• 日志格式：不同的日志配置会导致格式差异，需根据实际日志格式调整解析逻辑。
• 性能优化：若日志文件过大，建议按时间分片查询，避免一次性下载大文件。

内容的提问来源于stack exchange，提问作者User