如何在Cloud Storage中设置短于1天的对象生命周期（如5分钟）？

GCS原生的生命周期配置确实不支持小于1天的AGE值，且规则生效检查周期为24小时左右，所以要实现分钟级自动删除，必须借助Google Cloud的其他服务组合，以下是已验证的两种可靠方案：

方案一：Cloud Functions + Cloud Tasks（推荐，事件驱动型） #

这个方案通过对象创建事件触发函数，将延迟删除任务提交到Cloud Tasks队列，确保5分钟后精准执行删除操作，避免了函数超时或终止的风险。

步骤： #

1. 启用必要API：在Google Cloud控制台启用Cloud Functions API、Cloud Tasks API、Cloud Storage API。
2. 创建Cloud Tasks队列：
   • 进入Cloud Tasks控制台，创建一个新队列，可按需设置重试策略。
3. 编写触发函数（对象创建事件）：
   该函数在对象上传到GCS时触发，向Cloud Tasks提交延迟5分钟的删除任务。以下是Python示例：

   import os
   from datetime import datetime, timedelta
   from google.cloud import tasks_v2
   from google.protobuf import timestamp_pb2
   
   client = tasks_v2.CloudTasksClient()
   PROJECT_ID = os.environ.get("PROJECT_ID")
   LOCATION = os.environ.get("LOCATION")
   QUEUE_ID = os.environ.get("QUEUE_ID")
   DELETE_FUNCTION_URL = os.environ.get("DELETE_FUNCTION_URL")
   
   def schedule_delete(event, context):
       object_metadata = event
       bucket_name = object_metadata["bucket"]
       object_name = object_metadata["name"]
       
       # 计算5分钟后的执行时间
       schedule_time = datetime.utcnow() + timedelta(minutes=5)
       timestamp = timestamp_pb2.Timestamp()
       timestamp.FromDatetime(schedule_time)
   
       # 构建任务请求
       task = {
           "http_request": {
               "http_method": tasks_v2.HttpMethod.POST,
               "url": DELETE_FUNCTION_URL,
               "body": f'{{"bucket": "{bucket_name}", "object": "{object_name}"}}'.encode(),
               "headers": {"Content-Type": "application/json"}
           },
           "schedule_time": timestamp
       }
   
       # 提交任务到队列
       parent = client.queue_path(PROJECT_ID, LOCATION, QUEUE_ID)
       client.create_task(request={"parent": parent, "task": task})
   

   • 配置函数环境变量：PROJECT_ID、LOCATION、QUEUE_ID、DELETE_FUNCTION_URL（即下方删除函数的HTTP触发地址）。
   • 设置函数触发事件为google.storage.object.finalize，指定目标桶。
4. 编写删除对象的HTTP函数：
   该函数接收Cloud Tasks的请求，执行GCS对象删除操作。Python示例：

   from google.cloud import storage
   import json
   
   storage_client = storage.Client()
   
   def delete_object(request):
       request_data = json.loads(request.data)
       bucket_name = request_data["bucket"]
       object_name = request_data["object"]
       
       bucket = storage_client.bucket(bucket_name)
       blob = bucket.blob(object_name)
       blob.delete()
       
       return f"Deleted object {object_name} from {bucket_name}", 200
   

5. 配置权限：
   • 确保触发函数的服务账号拥有cloudtasks.tasks.create权限。
   • 确保Cloud Tasks队列的服务账号拥有调用删除函数的权限（可在函数权限设置中添加该账号为Cloud Functions Invoker）。
   • 确保删除函数的服务账号拥有storage.objects.delete权限。

方案二：Cloud Scheduler + 自定义扫描脚本（定时扫描型） #

如果不需要事件驱动，可通过定时任务定期扫描桶内创建时间超过5分钟的对象并删除，适合批量处理场景。

步骤： #

1. 启用Cloud Scheduler API：在控制台启用该API。
2. 编写扫描删除脚本：
   以下是Python示例，可部署为Cloud Function或Cloud Run服务：

   from google.cloud import storage
   from datetime import datetime, timedelta
   
   storage_client = storage.Client()
   TARGET_BUCKET = "your-bucket-name"
   DELETE_AFTER_MINUTES = 5
   
   def scan_and_delete(request):
       bucket = storage_client.bucket(TARGET_BUCKET)
       blobs = bucket.list_blobs()
       
       cutoff_time = datetime.utcnow() - timedelta(minutes=DELETE_AFTER_MINUTES)
       
       for blob in blobs:
           # 获取对象创建时间并做时区转换
           create_time = blob.time_created.replace(tzinfo=None)
           if create_time < cutoff_time:
               blob.delete()
               print(f"Deleted {blob.name}")
       
       return "Scan and delete completed", 200
   

3. 创建Cloud Scheduler任务：
   • 设置任务频率为*/5 * * * *（每5分钟执行一次）。
   • 任务目标为上述脚本的HTTP触发URL（如果是Cloud Function）或Cloud Run服务地址。
4. 配置权限：确保Scheduler的服务账号拥有调用脚本的权限，脚本的服务账号拥有GCS的对象列表和删除权限。

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