如何通过AWS SDK Python查询S3 Deep Archive恢复进度及剩余时间？

核心结论 #

AWS S3本身不提供恢复任务的实时进度百分比或精确剩余时间，通过head_object仅能获取恢复状态、请求时间、恢复层级等元数据，这是S3服务的设计限制——恢复操作是异步后台任务，AWS未暴露这类细粒度进度指标。

替代解决方案 #

1. 基于恢复层级的预估时间计算

根据AWS官方给出的不同恢复层级的时间范围，结合恢复请求发起时间，可大致估算剩余时间：

• Expedited：1-5分钟（仅支持≤250GB的对象）
• Standard：3-5小时
• Bulk：5-12小时

示例代码（结合x-amz-restore-request-date计算）：

from datetime import datetime, timezone
import time

object_key = 'deep-archive/00_00000.json'
bucket = 'your-bucket-name'

response = s3.head_object(Bucket=bucket, Key=object_key)
# 解析恢复请求发起时间
restore_request_date = datetime.strptime(
    response['RestoreRequestDate'], 
    '%a, %d %b %Y %H:%M:%S %Z'
).replace(tzinfo=timezone.utc)
current_time = datetime.now(timezone.utc)
elapsed_time = current_time - restore_request_date

# 根据恢复层级设定预估总时长（取官方范围中间值）
restore_tier = response['RestoreTier']
if restore_tier == 'Expedited':
    estimated_total_mins = 3  # 取1-5分钟的中间值
elif restore_tier == 'Standard':
    estimated_total_hours = 4  # 取3-5小时的中间值
elif restore_tier == 'Bulk':
    estimated_total_hours = 8.5  # 取5-12小时的中间值
else:
    estimated_total_hours = 0

# 计算剩余时间
if restore_tier == 'Expedited':
    remaining_mins = max(0, estimated_total_mins - elapsed_time.total_seconds()/60)
    print(f"已耗时: {elapsed_time.total_seconds()/60:.2f} 分钟，预估剩余时间: {remaining_mins:.2f} 分钟")
else:
    remaining_hours = max(0, estimated_total_hours - elapsed_time.total_seconds()/3600)
    print(f"已耗时: {elapsed_time.total_seconds()/3600:.2f} 小时，预估剩余时间: {remaining_hours:.2f} 小时")

注意：这只是基于官方预估的大致计算，实际恢复时间可能因对象大小、系统负载等因素波动。

2. 优化轮询检测恢复完成状态

继续使用head_object轮询，但根据恢复层级合理设置轮询间隔，避免频繁请求触发API速率限制：

import time

object_key = 'deep-archive/00_00000.json'
bucket = 'your-bucket-name'
# 根据恢复层级设置轮询间隔（秒）
poll_intervals = {
    'Expedited': 60,    # 1分钟一次
    'Standard': 300,    # 5分钟一次
    'Bulk': 600         # 10分钟一次
}

# 先获取恢复层级
initial_response = s3.head_object(Bucket=bucket, Key=object_key)
restore_tier = initial_response['RestoreTier']
poll_interval = poll_intervals.get(restore_tier, 300)

while True:
    response = s3.head_object(Bucket=bucket, Key=object_key)
    restore_status = response.get('Restore', '')
    if 'ongoing-request="false"' in restore_status:
        print("恢复完成")
        break
    print(f"恢复中... 下次检测将在 {poll_interval/60} 分钟后进行")
    time.sleep(poll_interval)

3. 使用S3事件通知被动接收完成通知

配置S3事件通知，当对象恢复完成时自动触发Lambda、SQS或SNS，无需主动轮询：

# 配置S3事件通知，将恢复完成事件发送到SNS主题
s3.put_bucket_notification_configuration(
    Bucket=bucket,
    NotificationConfiguration={
        'TopicConfigurations': [
            {
                'TopicArn': 'arn:aws:sns:your-region:your-account-id:your-sns-topic',
                'Events': ['s3:ObjectRestoreCompleted'],
                'Filter': {
                    'Key': {
                        'FilterRules': [
                            {'Name': 'prefix', 'Value': 'deep-archive/'}
                        ]
                    }
                }
            }
        ]
    }
)

配置完成后，当对象恢复完成时，你会收到对应的事件通知，无需持续轮询状态。

注意事项 #

• 避免过于频繁调用head_object，否则可能触发AWS API速率限制，导致请求被限流。
• 恢复完成后，对象会在x-amz-restore-expiry-days指定的天数内保持可访问状态，到期后自动回到Deep Archive存储类。

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