本文适用于使用火山引擎内容分发网络（CDN） 对 COS 存储桶中的文件进行分发加速的场景，介绍了如何使用腾讯云的云函数（SCF）服务，实现在腾讯云对象存储（COS）中的文件发生变更时自动触发火山引擎 CDN 对该文件进行刷新或预热。

场景介绍 #

当存储在腾讯云 COS 存储桶中的文件发生变更时，您可以通过腾讯云函数自动触发火山引擎 CDN 的缓存刷新或文件预热任务。文件更新的场景对应缓存刷新，即主动将 CDN 节点上缓存的资源标记为过期。新文件上传的场景对应文件预热，即将源站的资源主动缓存到 CDN 节点。

虽然火山引擎 CDN 提供了手动的缓存刷新和文件预热功能，但通过结合使用腾讯云函数，您可以将这一过程自动化，从而提高运维效率。

前提条件 #

在开始操作之前，请确保您已满足以下条件：

• 火山引擎 CDN
  * 已将腾讯云 COS 存储桶配置为加速域名的源站。
  * 已获取火山引擎 API 访问密钥，包含 Access Key ID 和 Secret Access Key。请确保该密钥拥有调用 SubmitRefreshTask 接口进行缓存刷新和调用 SubmitPreloadTask 接口进行文件预热的权限。关于如何创建和授权密钥，请参见 Access Key（密钥）管理 和 创建用户并授权。
• 腾讯云服务
  * 已开通腾讯云云函数（SCF）服务。

使用限制 #

• 该方案仅支持对文件 URL 提交刷新或预热任务，不支持目录刷新。
• 单函数单任务：为了确保函数逻辑清晰且易于维护，建议为刷新和预热任务分别创建独立的云函数。本文提供的代码示例遵循了该原则，分别为刷新和预热提供了独立的函数实现。
• CDN 配额：火山引擎 CDN 对缓存刷新和 URL 预热的提交频率及数量设有配额限制。在生产环境中使用此方案前，请务必查阅 刷新配额 和 预热配额 文档，以了解详细的配额信息，并评估您的业务场景是否满足配额要求。

操作步骤 #

步骤一：创建云函数 #

为了实现自动触发刷新或预热任务，您需要创建两个独立的云函数：一个用于刷新，另一个用于预热。以下步骤为创建刷新函数，预热函数的创建步骤类似。

1. 登录 腾讯云函数控制台。

   说明

   本文所需的函数代码见 函数代码。

2. 在左侧导航栏中，选择 函数服务。
3. 在 函数服务 页面，点击 新建。
4. 在 新建函数 页面，选择 从头开始 创建函数，并进行以下配置：
   • 函数类型：选择 事件函数。
   • 运行环境：选择 Python 3.9。
   • 为函数设置一个自定义名称，例如 volc-cdn-refresh。
   • 选择函数的部署地域。该地域必须与存储桶所在地域相同。
5. 勾选 我已阅读并同意《腾讯云云函数网络服务协议》。
6. 点击 完成。

步骤二：配置函数的环境变量和依赖 #

函数创建后，需要配置环境变量并安装依赖库。

1. 在函数详情页面，点击 函数管理 > 函数配置。

2. 点击 编辑，进入函数配置页面。

3. 在 环境变量 部分，配置以下用于火山引擎身份认证的变量：

   • Key: VOLC_SECRET_AK
   • Value: 您的火山引擎 Access Key ID
   • Key: VOLC_SECRET_SK
   • Value: 您的火山引擎 Secret Access Key

   说明

   代码中使用的 volcengine-python-sdk 会自动从环境变量中读取这些值用于 API 调用。

4. 点击 函数代码。

5. 在菜单栏中，点击 终端 > 新建终端。

6. 在 终端 中执行以下命令来安装函数所需的 volcengine-python-sdk 库：

   cd src
   sudo pip install volcengine-python-sdk -t .
   

7. 安装完成后，点击 函数配置 标签页，然后点击页面底部的 保存。

步骤三：部署和测试函数 #

配置完成后，您需要部署函数并进行测试，以确保函数能够正确执行。

1. 点击 函数代码 标签页。
2. 将本文 附录：函数代码 中对应的函数代码粘贴至代码编辑器中，并根据注释提示替换为您自己的 CDN 加速域名。
3. 在页面下方，点击 部署。
4. 部署成功后，设置 测试事件 为 COS 对象存储的 PUT 事件模板。
5. 点击 测试。

在执行日志中，如果出现以下信息，表示函数已成功执行。

• 刷新任务提交成功，任务 ID：xxxxxxxxxxx。

在 火山引擎 CDN 控制台 的 刷新预热 > 操作记录 页面，您会看到手动触发的任务。

步骤四：配置 COS 触发器 #

为了在 COS 存储桶中的文件发生变更时自动触发函数，您需要为函数配置一个 COS 触发器。

1. 在函数的 函数管理 页面，点击 触发管理。
2. 点击 创建触发器。
3. 在 创建触发器 页面，配置以下参数：
   • 触发方式：选择 COS 触发。
   • COS Bucket：选择您的存储桶。
   • 事件类型：选择一个适配您业务场景的事件类型。参见 COS 触发器说明。
4. 点击 提交。

步骤五：验证 #

配置完成后，您可以通过上传或更新文件到您的 COS 存储桶来验证整个流程是否正常工作。

1. 上传一个新文件或更新一个现有文件到您的 COS 存储桶。
2. 在 火山引擎 CDN 控制台 的 刷新预热 > 操作记录 页面，查看刷新或预热任务的执行状态。

注意事项 #

API 调用频率与函数并发：为避免函数的并发执行导致调用火山引擎 API 超出频率限制，您需要为函数配置合理的并发额度。火山引擎 CDN 的刷新和预热 API 每秒最多可以处理 20 个请求。如果您有批量上传文件的场景，建议在函数的 并发配额 页面，设置 函数最大独占配额 为3，以匹配 CDN 的 API 调用限制。

附录：函数代码 #

您可以根据需要，选择使用刷新函数或预热函数。在创建云函数后，将对应的代码粘贴到函数代码编辑器中。

刷新函数 #

from __future__ import print_function
import volcenginesdkcore
import volcenginesdkcdn
from volcenginesdkcore.rest import ApiException
import json
import string
import sys
import logging
import os
import uuid

# pip install volcengine-python-sdk
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger()
logger.setLevel(level=logging.INFO)

def main_handler(event, context):
    # 格式见 https://cloud.tencent.com/document/product/583/9707
    logger.info("start main handler")
    if "Records" not in event.keys():
        return {"code": 410, "errorMsg": "event is not come from cos"}
    cos_event = event['Records'][0]['cos']
    appid = cos_event['cosBucket']['appid']
    bucket = cos_event['cosBucket']['name']
    key = cos_event['cosObject']['key']  # 获取更新的文件路径
    key = key.replace('/' + str(appid) + '/' + bucket + '/', '', 1) 
 
    secret_ak = os.environ.get('VOLC_SECRET_AK')      
    secret_sk = os.environ.get('VOLC_SECRET_SK')   

    cdn_domain = "https://www.example.com"  # 替换为实际火山 CDN 域名
    refresh_url = cdn_domain + '/' + key
    logger.info("refresh_url is " + refresh_url)

    configuration = volcenginesdkcore.Configuration()
    configuration.ak = "secret_ak"  # 火山 CDN AK
    configuration.sk = "secret_sk"  # 火山 CDN SK
    configuration.region = "cn-north-1"
    volcenginesdkcore.Configuration.set_default(configuration)
    api_instance = volcenginesdkcdn.CDNApi()

    submit_refresh_task_request = volcenginesdkcdn.SubmitRefreshTaskRequest(
        type="file",
        urls=refresh_url
    )
    try:
        api_response = api_instance.submit_refresh_task(submit_refresh_task_request)
        logger.info(api_response)
        print(f"刷新任务提交成功，任务ID：{api_response.task_id}")
    except ApiException as e:
        print(f"刷新任务提交失败，错误信息：{e}")
    
    return {"status": "success"}

预热函数 #

from __future__ import print_function
import volcenginesdkcore
import volcenginesdkcdn
from volcenginesdkcore.rest import ApiException
import json
import string
import sys
import logging
import os
import uuid

# pip install volcengine-python-sdk
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger()
logger.setLevel(level=logging.INFO)

def main_handler(event, context):
    # 格式见 https://cloud.tencent.com/document/product/583/9707
    logger.info("start main handler")
    if "Records" not in event.keys():
        return {"code": 410, "errorMsg": "event is not come from cos"}
    cos_event = event['Records'][0]['cos']
    appid = cos_event['cosBucket']['appid']
    bucket = cos_event['cosBucket']['name']
    key = cos_event['cosObject']['key']  # 获取更新的文件路径
    key = key.replace('/' + str(appid) + '/' + bucket + '/', '', 1) 
 
    secret_ak = os.environ.get('VOLC_SECRET_AK')      
    secret_sk = os.environ.get('VOLC_SECRET_SK')   

    cdn_domain = "https://www.example.com"  # 替换为实际火山 CDN 域名
    preload_url = cdn_domain + '/' + key
    logger.info("preload_url is " + preload_url)

    configuration = volcenginesdkcore.Configuration()
    configuration.ak = "secret_ak"  # 火山 CDN AK
    configuration.sk = "secret_sk"  # 火山 CDN SK
    configuration.region = "cn-north-1"
    volcenginesdkcore.Configuration.set_default(configuration)
    api_instance = volcenginesdkcdn.CDNApi()

    submit_preload_task_request = volcenginesdkcdn.SubmitPreloadTaskRequest(
        urls=preload_url
    )
    try:
        api_response = api_instance.submit_preload_task(submit_preload_task_request)
        logger.info(api_response)
        print(f"预热任务提交成功，任务ID：{api_response.task_id}")
    except ApiException as e:
        print(f"预热任务提交失败，错误信息：{e}")
    
    return {"status": "success"}