华为云 FunctionGraph 实战：用 Serverless 函数自动清理 OBS 过期文件

在对象存储（OBS）的实际使用中，我们经常会遇到一个头疼的问题：临时文件、日志文件、备份文件不断堆积，桶里的对象越来越多，存储费用也跟着水涨船高。手动清理既繁琐又容易遗漏，用一台 ECS 跑定时脚本又觉得"杀鸡用牛刀"——为了清理文件专门养一台服务器，成本不划算。

今天分享一个轻量方案：用华为云 FunctionGraph（函数工作流）+ 定时触发器，实现 OBS 过期文件的自动清理。整个方案不需要维护任何服务器，函数只在执行时才计费，非常适合这种低频定时任务。

一、方案思路

整体架构非常简单：

1. 在 OBS 桶中，按目录前缀存放文件，比如 logs/2026-11/ 存放日志，backup/ 存放临时备份。
2. 在 FunctionGraph 上创建一个 Python 函数，函数逻辑是：列出指定前缀下的所有对象，对比每个对象的最后修改时间，超过保留天数的就删除。
3. 给函数挂一个定时触发器（TIMER），每天凌晨执行一次。

这样就能实现"每天自动扫一遍，过期的文件自动删掉"，全程无人值守。

二、函数代码实现

函数运行时选 Python 3.9，代码逻辑如下：

import json
import time
from datetime import datetime, timedelta

# 华为云 OBS Python SDK
from obs import ObsClient

# 硬编码 AK/SK 仅用于演示，生产环境请通过环境变量注入
AK = 'your-access-key'
SK = 'your-secret-key'
SERVER = 'obs.cn-north-4.myhuaweicloud.com'
BUCKET_NAME = 'my-cleanup-bucket'

# 初始化 OBS 客户端（在函数外部初始化，可复用连接）
obs_client = ObsClient(access_key_id=AK, secret_access_key=SK, server=SERVER)

def handler(event, context):
    """
    定时触发入口。
    event 中可携带配置，也可从环境变量读取。
    默认清理 logs/ 前缀下超过 7 天的文件。
    """
    # 从事件或环境变量获取参数，支持灵活配置
    prefix = event.get('prefix', 'logs/')
    retain_days = int(event.get('retain_days', 7))

    # 计算过期时间戳（秒）
    now = datetime.utcnow()
    expire_before = now - timedelta(days=retain_days)
    expire_ts = expire_before.timestamp()

    print(f"开始清理: bucket={BUCKET_NAME}, prefix={prefix}, "
          f"保留{retain_days}天, 过期边界={expire_before.isoformat()}")

    deleted_count = 0
    scanned_count = 0
    marker = None

    while True:
        # 分页列举对象，每页最多 1000 个
        resp = obs_client.listObjects(
            BUCKET_NAME, prefix=prefix, marker=marker, max_keys=1000
        )

        if resp.status < 300 and resp.body.contents:
            for obj in resp.body.contents:
                scanned_count += 1
                # 跳过"目录占位符"（以 / 结尾的空对象）
                if obj.key.endswith('/'):
                    continue

                # 对比最后修改时间
                last_modified = float(obj.lastModified) / 1000.0
                if last_modified < expire_ts:
                    del_resp = obs_client.deleteObject(BUCKET_NAME, obj.key)
                    if del_resp.status < 300:
                        deleted_count += 1
                        print(f"已删除: {obj.key} "
                              f"(修改时间: {datetime.utcfromtimestamp(last_modified).isoformat()})")
                    else:
                        print(f"删除失败: {obj.key}, "
                              f"errorCode={del_resp.errorCode}, "
                              f"errorMsg={del_resp.errorMessage}")

            # 翻页
            if resp.body.isTruncated:
                marker = resp.body.nextMarker
            else:
                break
        else:
            print(f"列举对象失败: status={resp.status}, "
                  f"errorCode={resp.errorCode}")
            break

    result = {
        'scanned': scanned_count,
        'deleted': deleted_count,
        'prefix': prefix,
        'retain_days': retain_days
    }
    print(f"清理完成: {json.dumps(result, ensure_ascii=False)}")
    return result

代码逻辑拆解：

• 分页列举：OBS listObjects 每次最多返回 1000 个对象，用 while 循环 + marker 翻页，确保不遗漏。
• 跳过目录占位符：OBS 中没有真正的"目录"，以 / 结尾的空对象是前端控制台用来模拟目录结构的，不需要删。
• 时间对比：把对象的 lastModified（毫秒时间戳）转成秒，和过期边界比较，早于边界就删。
• 日志输出：每次删除都打印一条日志，方便在 FunctionGraph 控制台的"日志"页签回溯执行记录。

三、配置定时触发器

函数写好后，在 FunctionGraph 控制台给它添加一个 TIMER 触发器。触发器支持两种格式：

方式一：Rate 表达式（简单间隔）

Rate(1 days)

表示每 1 天执行一次。适合"每天扫一次"这种简单场景。

方式二：Cron 表达式（精确定时）

0 2 1/1 * ? *

这是 Quartz Cron 格式，含义是：每天凌晨 2 点执行。六个字段依次是：秒、分、时、日、月、周、年。注意华为云的 Cron 用的是 Quartz 风格，不是 Linux crontab 风格，日字段和周字段不能同时用 *，其中一个要用 ?。

我推荐用 Cron 方式，把清理任务安排在凌晨低峰期，避免和其他业务高峰撞车。

四、多桶多前缀的扩展

实际场景中，一个桶里可能有多种类型的文件，保留策略各不相同。比如：

• logs/ 下的日志保留 7 天
• backup/ 下的临时备份保留 3 天
• report/ 下的日报保留 30 天

有两种做法：

做法 A：一个函数，事件传参

给函数配多个触发器，每个触发器的事件体传入不同的 prefix 和 retain_days：

{"prefix": "logs/", "retain_days": 7}

{"prefix": "backup/", "retain_days": 3}

函数代码不用改，靠事件参数区分。但 TIMER 触发器的事件体是固定配置的，改起来要去控制台编辑触发器，不够灵活。

做法 B：一个函数，内部循环

在函数里维护一个策略列表，一次执行遍历所有前缀：

CLEANUP_POLICIES = [
    {'prefix': 'logs/', 'retain_days': 7},
    {'prefix': 'backup/', 'retain_days': 3},
    {'prefix': 'report/', 'retain_days': 30},
]

def handler(event, context):
    for policy in CLEANUP_POLICIES:
        cleanup_prefix(policy['prefix'], policy['retain_days'])

我更推荐做法 B，策略集中管理，加一条规则改一行代码就行，而且一次函数执行处理所有前缀，减少函数调用次数，省钱。

五、几个踩坑经验

1. 函数执行超时

默认超时时间是 3 秒，对于对象多的桶根本不够。建议根据桶内对象数量估算：每 1000 个对象大约需要 2-3 秒（列举 + 逐个删除）。桶里如果有几万个对象，把超时设到 60 秒甚至 300 秒。在函数配置页面的"超时时间"那里改。

2. 函数内存配额

清理任务主要是 IO 密集型（列举 + 删除 API 调用），CPU 消耗很低，128MB 内存足够。不需要调大内存，省下来的配额留给其他函数。

3. IAM 委托授权

函数要操作 OBS，需要 OBS 的读写权限。最干净的做法是在函数配置里关联一个 IAM 委托（委托），委托里赋予 OBS桶的读写策略。不要在代码里硬编码 AK/SK——虽然演示代码里我这么写了，但生产环境请务必用委托方式，AK/SK 泄露的风险远大于多花五分钟配置委托。

4. 删除是不可逆的

OBS deleteObject 执行后对象直接消失，不进回收站（除非桶开了多版本控制）。建议先在测试桶上跑几遍，确认清理逻辑无误后再挂到生产桶。或者在函数里先加一个 dry_run 参数，为 True 时只打印"将要删除"的列表，不真正执行删除，人工确认后再关掉 dry_run。

六、成本估算

这个方案的成本几乎可以忽略：

• FunctionGraph：每月 30 次定时触发，每次执行几秒，免费额度（每月 100 万次请求、400,000 GB-秒）完全覆盖。
• OBS：列举和删除 API 调用，每万次请求 0.1 元，一个月几十次调用，几乎为零。
• 节省的存储费：假设每天清理 10GB 过期文件，标准存储 0.099 元/GB/月，每月省下约 30 元。

用零点几毛钱的函数调用费，换掉几十块的无效存储费，这笔账很划算。

七、总结

整个方案的核心就三步：写函数 → 挂定时触发器 → 配好权限。不维护服务器，不跑常驻进程，函数按需执行，用完即释。这种"事件驱动 + Serverless"的模式，天然适合所有低频、定时的运维场景——不只是清理 OBS，清理临时缓存、轮转备份、生成日报，思路都是一样的。

希望这篇分享能给正在为 OBS 存储费用发愁的朋友一个参考。如果你有更好的方案或者踩过其他坑，欢迎在评论区交流。