文档中心

如何使用函数服务实现对象存储同步刷新CDN

最近更新时间：2023.05.17 16:48:05

首次发布时间：2023.05.17 16:48:05

前言

本实验使用函数服务，实现对象存储资源删除、上传时自动调用刷新CDN缓存的接口。

关于实验

预计部署时间：20分钟
级别：初级
相关产品：函数服务、对象存储、内容分发网络
受众：通用

实验说明

点击此链接登录控制台。
如果您还没有账户，请点击此链接注册账户。

实验步骤

第一步创建函数服务

1.登录函数服务控制台。

2.函数列表页面，单击创建函数。
alt
3.创建函数。
这里选择基于Hello World 模版创建即可。
alt
4.函数配置。
alt
调用cdn刷新缓存的接口代码参考如下：

import datetime
import hashlib
import hmac
import json
from urllib.parse import quote
import requests


Service = "CDN"
Version = "2021-03-01"
Region = "cn-north-1"
Host = "cdn.volcengineapi.com"
AK = ""
SK = ""
domain = ""


def norm_query(params):
    query = ""
    for key in sorted(params.keys()):
        if type(params[key]) == list:
            for k in params[key]:
                query = (
                        query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&"
                )
        else:
            query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&")
    query = query[:-1]
    return query.replace("+", "%20")


# 第一步：准备辅助函数。
# sha256 非对称加密
def hmac_sha256(key: bytes, content: str):
    return hmac.new(key, content.encode("utf-8"), hashlib.sha256).digest()


# sha256 hash算法
def hash_sha256(content: str):
    return hashlib.sha256(content.encode("utf-8")).hexdigest()


# 第二步：创建一个 CDN 的 API 请求函数。签名计算的过程包含在该函数中。
def request(method, query, header, ak, sk, action, body):
    # 第三步：创建身份证明。其中的 Service 和 Region 字段是固定的。ak 和 sk 分别代表
    # AccessKeyID 和 SecretAccessKey。同时需要初始化签名结构体。一些签名计算时需要的属性也在这里处理。
    # 初始化身份证明结构体
    credential = {
        "access_key_id": ak,
        "secret_access_key": sk,
        "service": Service,
        "region": Region,
    }
    # 初始化签名结构体
    request_param = {
        "body": json.dumps(body),
        "host": Host,
        "path": "/",
        "method": method,
        "content_type": "application/json",
        "date": datetime.datetime.utcnow(),
        "query": {"Action": action, "Version": Version, **query},
    }
    # 第四步：接下来开始计算签名。在计算签名前，先准备好用于接收签算结果的 signResult 变量，并设置一些参数。
    # 初始化签名结果的结构体
    x_date = request_param["date"].strftime("%Y%m%dT%H%M%SZ")
    short_x_date = x_date[:8]
    x_content_sha256 = hash_sha256(request_param["body"])
    sign_result = {
        "Host": request_param["host"],
        "X-Content-Sha256": x_content_sha256,
        "X-Date": x_date,
        "Content-Type": request_param["content_type"],
    }
    # 第五步：计算 Signature 签名。
    signed_headers_str = ";".join(
        ["content-type", "host", "x-content-sha256", "x-date"]
    )
    canonical_request_str = "\n".join(
        [request_param["method"],
         request_param["path"],
         norm_query(request_param["query"]),
         "\n".join(
             [
                 "content-type:" + request_param["content_type"],
                 "host:" + request_param["host"],
                 "x-content-sha256:" + x_content_sha256,
                 "x-date:" + x_date,
             ]
         ),
         "",
         signed_headers_str,
         x_content_sha256,
         ]
    )
    hashed_canonical_request = hash_sha256(canonical_request_str)
    credential_scope = "/".join([short_x_date, credential["region"], credential["service"], "request"])
    string_to_sign = "\n".join(["HMAC-SHA256", x_date, credential_scope, hashed_canonical_request])
    k_date = hmac_sha256(credential["secret_access_key"].encode("utf-8"), short_x_date)
    k_region = hmac_sha256(k_date, credential["region"])
    k_service = hmac_sha256(k_region, credential["service"])
    k_signing = hmac_sha256(k_service, "request")
    signature = hmac_sha256(k_signing, string_to_sign).hex()
    sign_result["Authorization"] = "HMAC-SHA256 Credential={}, SignedHeaders={}, Signature={}".format(
        credential["access_key_id"] + "/" + credential_scope,
        signed_headers_str,
        signature,
    )
    header = {**header, **sign_result}
    # 第六步：将 Signature 签名写入 HTTP Header 中，并发送 HTTP 请求。
    r = requests.post("https://{}{}".format(request_param["host"], request_param["path"]),
                      headers=header,
                      params=request_param["query"],
                      data=request_param["body"],
                      )
    return r.json()


def handler(event, context):
    # Log incoming request.
	#
	# NOTE: the log here is only for debug purpose. It's recommended to delete those debug/info
	# logs and only print log where error occurs after your business logic is verified and ready to
	# go production, nor the log amount may be too huge.
    print(f"received new request, event content: {event}")

    key = event['data']['events'][0]['tos']['object']['key']   
    print(key) 
    url = 'http://'+ domain + '/' + key
    print(url)
    request_body = {
        "Type": "file",
        "Urls": url
    }
    response_body = request("POST", {}, {}, AK, SK, "SubmitRefreshTask", request_body)
    print(response_body)

    result = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        },
        'body': json.dumps({
            'message': 'submit refresh task success!'
        })
    }
    return result

5.发布版本。
alt
6.等待发布完成，创建TOS触发器。
这里以触发事件选择 tos:Object:Created:* tos:Object:Created:* 为例,当检测到对象上传以及删除的时候，会触发函数执行。
alt
alt