腾讯云轻量应用服务器创建云函数来实现自动快照
腾讯云的轻量应用服务器虽然提供了快照功能但却不能设置自动快照,只能手动执行非常麻烦。但是我们可以主动调用腾讯云的 api 接口来实现快照功能,结合上腾讯云的云函数功能就可以实现定时自动快照了。
现在分两个板块进行:
- 通过 Python + 腾讯云 SDK 本地实现创建快照功能。
- 将本地实现的功能移动到腾讯云云函数上从而实现自动快照。
在本地直接调用 Python SDK 来实现创建快照
从腾讯云控制台中获取必要信息
如果我们要手动调用 SDK 创建快照的话至少需要这些信息:
- 用户密钥:也就是
secretId和secretKey,这两个东西需要我们手动在控制台生成用于用户认证。- 地域:你的服务器所在地域
- 实例ID:你的服务器实例ID 我们现在依次介绍这些信息的获取方式。
创建子账号并获取用户密钥
对于 API 密钥的获取,我们使用腾讯云的推荐方式,创建一个子账号授予编程访问方式和轻量应用服务器全部权限。这个用户专门用于创建快照等操作,避免授予过多权限。
登录腾讯云控制台,搜索“访问管理”,第一个结果就是。

点击侧边栏的用户列表,然后点击“新建用户”,点击“快速创建”

访问方式只需要勾选编程访问

用户权限搜索“轻量应用服务器”然后勾选“QcloudLighthouseFullAccess”

创建完成后就自动弹出并展示一个默认创建的 secretId 和 secretKey,记录下来(如果没有弹出就自行进入下面这里创建)

获取地域代码和实例 ID
进入腾讯云控制台,轻量应用服务器,在概要里面可以找到地域和实例ID,记录下来。

地域需要进行一个转换,我这里是广州,直接搜索腾讯云地域列表就可以查找的类似下面这种,很容易找到,比如我的地域是广州,就是记下地域 ID: ap-guangzhou

创建 Python 项目并导入腾讯云 SDK
- 创建一个 Python 项目命名为
tencent-cloud-scf,这里我们使用 uv 作为包管理器。
uv init tencent-cloud-scf
cd tencent-cloud-scf- 从腾讯云的 SDK 中心我们可以找到轻量应用服务器的 Python SDK

在其中搜索,第二个就是我们要的 SDK 了,我们可以看到包名是 tencentcloud-sdk-python-lighthouse ,接着我们将 SDK 作为项目依赖引入。
uv add tencentcloud-sdk-python-lighthouse编写Python 代码实现创建快照功能
touch index.py
vim index.py将如下内容拷贝到 index.py (以下代码依赖 ai 编写完成,不保证质量如何,但是经过实际验证功能上没有问题):
以下脚本实现的功能主要是创建快照并限制保留的快照数量,请仔细阅读代码了解环境变量的含义
# -*- coding: utf-8 -*-
"""腾讯云 SCF 入口:按保留数量自动创建轻量应用服务器快照。"""
import json
import os
import time
import traceback
from dataclasses import dataclass
from datetime import datetime, timezone
from tencentcloud.common.credential import Credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.lighthouse.v20200324.lighthouse_client import LighthouseClient
from tencentcloud.lighthouse.v20200324.models import (
CreateInstanceSnapshotRequest,
DeleteSnapshotsRequest,
DescribeSnapshotsRequest,
Filter,
)
# 需要配置的环境内容:
#
# secretId
ENV_SECRET_ID = "SECRET_ID"
# secretKey
ENV_SECRET_KEY = "SECRET_KEY"
# 地域 ID
ENV_REGION = "REGION"
# 服务器实例ID
ENV_INSTANCE_ID = "LIGHTHOUSE_INSTANCE_ID"
# 保留的最大快照数量,如果超过这个数量就删除最老的快照然后创建新的(腾讯云默认情况下一台机器2个配额);未配置时为 `2`
ENV_MAX_SNAPSHOT_COUNT = "MAX_SNAPSHOT_COUNT"
# 等待单个快照删除完成的最长秒数;未配置时为 `60`
ENV_DELETE_TIMEOUT_SECONDS = "SNAPSHOT_DELETE_TIMEOUT_SECONDS"
DFAULT_REGION = "ap-guangzhou"
DEFAULT_MAX_SNAPSHOT_COUNT = 2
DEFAULT_DELETE_TIMEOUT_SECONDS = 60
SNAPSHOT_QUERY_LIMIT = 100
DELETE_POLL_INTERVAL_SECONDS = 2
# 创建的快照名前缀,这里没有设置为环境变量写死
SNAPSHOT_NAME_PREFIX = "auto-snapshot"
def _required_env(name: str) -> str:
"""读取必填环境变量,避免因缺少配置而以不明确的方式调用 SDK。"""
value = os.environ.get(name)
if not value:
raise ValueError(f"缺少环境变量:{name}")
return value
def _positive_int(value: object, name: str) -> int:
"""将事件或环境变量中的正整数配置转换为 int。"""
if isinstance(value, bool):
raise ValueError(f"{name} 必须是大于 0 的整数")
try:
number = int(value)
except (TypeError, ValueError) as error:
raise ValueError(f"{name} 必须是大于 0 的整数") from error
if number <= 0:
raise ValueError(f"{name} 必须是大于 0 的整数")
return number
@dataclass(frozen=True)
class FunctionConfig:
"""云函数运行所需的全部环境变量配置。"""
secret_id: str
secret_key: str
region: str
instance_id: str
max_snapshot_count: int
delete_timeout_seconds: int
@classmethod
def from_environment(cls) -> "FunctionConfig":
"""一次性读取并校验所有环境变量。"""
return cls(
secret_id=_required_env(ENV_SECRET_ID),
secret_key=_required_env(ENV_SECRET_KEY),
region=os.environ.get(ENV_REGION, DEFAULT_REGION),
instance_id=_required_env(ENV_INSTANCE_ID),
max_snapshot_count=_positive_int(
os.environ.get(ENV_MAX_SNAPSHOT_COUNT, str(DEFAULT_MAX_SNAPSHOT_COUNT)),
ENV_MAX_SNAPSHOT_COUNT,
),
delete_timeout_seconds=_positive_int(
os.environ.get(ENV_DELETE_TIMEOUT_SECONDS, str(DEFAULT_DELETE_TIMEOUT_SECONDS)),
ENV_DELETE_TIMEOUT_SECONDS,
),
)
def _get_max_snapshot_count(event: object, config: FunctionConfig) -> int:
"""优先使用调用事件中的 max_snapshot_count,否则使用环境变量默认值。"""
event_value = event.get("max_snapshot_count") if isinstance(event, dict) else None
return config.max_snapshot_count if event_value is None else _positive_int(event_value, "max_snapshot_count")
def _describe_instance_snapshots(client: LighthouseClient, instance_id: str) -> list[dict]:
"""仅查询目标实例的快照,绝不将其他实例的快照纳入删除范围。"""
instance_filter = Filter()
instance_filter.Name = "instance-id"
instance_filter.Values = [instance_id]
request = DescribeSnapshotsRequest()
request.Filters = [instance_filter]
request.Limit = SNAPSHOT_QUERY_LIMIT
result = json.loads(client.DescribeSnapshots(request).to_json_string())
return result.get("SnapshotSet", [])
def _snapshot_sort_key(snapshot: dict) -> str:
"""腾讯云返回 UTC ISO 8601 时间,字符串排序即可保持创建时间顺序。"""
created_time = snapshot.get("CreatedTime")
if not isinstance(created_time, str) or not created_time:
snapshot_id = snapshot.get("SnapshotId", "未知")
raise ValueError(f"快照 {snapshot_id} 缺少 CreatedTime,不能安全判断最旧快照")
return created_time
def _wait_until_deleted(client: LighthouseClient, config: FunctionConfig, snapshot_id: str) -> None:
"""确认删除完成后才创建新快照,避免与快照配额或异步操作发生竞争。"""
deadline = time.monotonic() + config.delete_timeout_seconds
while True:
snapshot_ids = {
snapshot.get("SnapshotId")
for snapshot in _describe_instance_snapshots(client, config.instance_id)
}
if snapshot_id not in snapshot_ids:
print(f"已确认删除快照:{snapshot_id}")
return
remaining_seconds = deadline - time.monotonic()
if remaining_seconds <= 0:
raise TimeoutError(
f"等待删除快照 {snapshot_id} 超时({config.delete_timeout_seconds} 秒)"
)
time.sleep(min(DELETE_POLL_INTERVAL_SECONDS, remaining_seconds))
def _delete_oldest_snapshots(
client: LighthouseClient,
config: FunctionConfig,
snapshots: list[dict],
max_snapshot_count: int,
) -> list[str]:
"""删除足够数量的最旧快照,为新快照预留一个位置。"""
delete_count = max(0, len(snapshots) - max_snapshot_count + 1)
deleted_snapshot_ids = []
for snapshot in sorted(snapshots, key=_snapshot_sort_key)[:delete_count]:
snapshot_id = snapshot.get("SnapshotId")
snapshot_state = snapshot.get("SnapshotState")
if not snapshot_id:
raise ValueError("快照缺少 SnapshotId,不能安全删除")
# DeleteSnapshots 仅允许删除 NORMAL 状态快照。
if snapshot_state != "NORMAL":
raise RuntimeError(
f"最旧快照 {snapshot_id} 当前状态为 {snapshot_state},"
"必须等待其变为 NORMAL 后才能删除和创建新快照"
)
request = DeleteSnapshotsRequest()
request.SnapshotIds = [snapshot_id]
delete_response = json.loads(client.DeleteSnapshots(request).to_json_string())
print(f"已提交删除最旧快照:{snapshot_id},请求 ID:{delete_response.get('RequestId', '未知')}")
_wait_until_deleted(client, config, snapshot_id)
deleted_snapshot_ids.append(snapshot_id)
return deleted_snapshot_ids
def main_handler(event, context):
"""为指定轻量应用服务器创建快照,并按数量上限自动淘汰最旧快照。
函数作用:每次调用均为 ``LIGHTHOUSE_INSTANCE_ID`` 指定的实例创建一个系统盘
快照。若该实例现有快照数达到保留上限,函数会先删除最旧且状态为 ``NORMAL``
的快照,确认删除完成后再创建新快照,避免误删其他实例的快照。
环境变量:
SECRET_ID(必填):拥有 Lighthouse 快照管理权限的 CAM SecretId。
SECRET_KEY(必填):对应的 CAM SecretKey。
LIGHTHOUSE_INSTANCE_ID(必填):目标实例 ID,例如 ``lhins-xxxxxxxx``。
REGION(可选):实例所在地域,默认 ``ap-guangzhou``。
MAX_SNAPSHOT_COUNT(可选):默认保留快照数量,默认 ``7``。
SNAPSHOT_DELETE_TIMEOUT_SECONDS(可选):等待单个删除操作完成的最大秒数,默认 ``60``。
调用事件:
可传入 ``{"max_snapshot_count": 7}`` 覆盖本次调用的保留数量;定时触发器
使用空事件 ``{}`` 时采用 ``MAX_SNAPSHOT_COUNT``。
返回值:包含目标实例 ID、实际保留上限、已删除快照 ID、新快照 ID 与腾讯云请求 ID。
"""
try:
config = FunctionConfig.from_environment()
max_snapshot_count = _get_max_snapshot_count(event, config)
print(
f"开始执行快照保留任务,实例:{config.instance_id},地域:{config.region},"
f"最大保留数量:{max_snapshot_count}"
)
credential = Credential(config.secret_id, config.secret_key)
client = LighthouseClient(credential, config.region, ClientProfile())
snapshots = _describe_instance_snapshots(client, config.instance_id)
print(f"当前实例已有 {len(snapshots)} 个快照")
deleted_snapshot_ids = _delete_oldest_snapshots(
client,
config,
snapshots,
max_snapshot_count,
)
create_request = CreateInstanceSnapshotRequest()
create_request.InstanceId = config.instance_id
create_request.SnapshotName = (
f"{SNAPSHOT_NAME_PREFIX}-{datetime.now(timezone.utc):%Y%m%d-%H%M%S}"
)
create_response = json.loads(client.CreateInstanceSnapshot(create_request).to_json_string())
result = {
"InstanceId": config.instance_id,
"MaxSnapshotCount": max_snapshot_count,
"DeletedSnapshotIds": deleted_snapshot_ids,
"SnapshotId": create_response.get("SnapshotId"),
"RequestId": create_response.get("RequestId"),
}
print(f"新快照创建请求已提交:\n{json.dumps(result, ensure_ascii=False, indent=2)}")
return result
except Exception as error:
print(f"快照保留任务失败:{type(error).__name__}: {error}")
traceback.print_exc()
raise
if __name__ == "__main__":
main_handler({}, None)设置临时环境变量并执行代码:
export SECRET_ID=xxx
export SECRET_KEY=xxx
export REGION=xxx
export LIGHTHOUSE_INSTANCE_ID=xxx
export MAX_SNAPSHOT_COUNT=xxx
uv run index.py验证成功创建了前缀为 auto-snapshot 的快照。
将本地代码迁移到腾讯云函数定时执行
我们已经在本地成功执行了快照创建,同时会限制快照数量,现在我们将这项功能迁移到腾讯的云函数,并设置定时触发从而实现自动快照功能。
- 我们将腾讯云的 SDK 下载打包压缩为一个 zip,后续将这个 zip 上传为“层”。
mkdir tencentcloud-layer && cd tencentcloud-layer
pip3 install tencentcloud-sdk-python -t .
zip -r tencentcloud-sdk-layer.zip *- 在腾讯云云函数控制台新建层(我们之前不是引入了腾讯云的 SDK 吗,我们将这个 SDK 创建为层,这样可以在不同的云函数里面引用层,相当于引用依赖)

3.创建一个云函数,选择“从头开始”

将函数代码拷贝进去。
注意这里不用把 main 函数拷贝进去,云函数默认执行的是
def main_handler(event, context)方法

最后设置触发器指定定时执行。

然后点进刚刚创建的云函数,在“层管理”这里将之前创建的层绑定到这个云函数(不然是无法识别依赖的)

在函数代码下面点击“部署”,然后点击“测试”,验证云函数是否正确执行:
