阿里云日志服务新增软删除(Soft Deletion) 功能,允许用户在删除日志数据时,将数据标记为“已删除”状态而非立即物理清除。本文主要介绍如何进行软删除。
日志服务软删除功能按照用户和地域灰度,如需要开启,请提工单申请。
背景信息
在数据驱动环境下,日志和可观测数据成为企业关键资产,但也带来了数据管理、合规性和安全性等挑战。为应对这些问题,阿里云日志服务(SLS)推出软删除(Soft Deletion)功能,提供可追溯的删除操作机制,为您的数据安全与合规体系提供的一项关键能力和一个强大的安全网。
工作原理(点击查看)
应用场景(点击查看)
注意事项
存储成本考量:软删除的数据在保留期内依然占用存储空间并产生费用。数据需要等Logstore过期才会真正物理删除。
查询语句的精确性:在删除前,先查询。在执行删除操作之前,务必使用完全相同的查询时间和查询语句执行一次查询操作,预览将要被删除的数据,确认无误后再执行删除。
软删除的日志不可查询分析(包括查询、SQL 分析、ScheduledSQL、告警功能),对流式处理操作不影响(包括消费、数据加工、投递等)。
软删除不可逆,日志删除了不可恢复。
支持地域
目前新加坡和华北6(乌兰察布)支持软删除。您可以在Project的概览页面,查看该Project所在地域的服务入口。各地域的Region ID,请参见服务入口。
权限须知
若您使用阿里云主账号登录,默认拥有所有操作权限,可直接对Project进行相关操作。
若您使用RAM账号登录,请根据需要向主账号使用者申请如下两种日志服务的系统策略。
系统权限策略:权限范围较大,用户无法修改系统权限策略的内容,但配置步骤简单。
AliyunLogFullAccess:管理日志服务的权限。
AliyunLogReadOnlyAccess:只读访问日志服务的权限。
自定义权限策略:强烈建议遵循最小权限原则,仅为受信任的人员或自动化服务授予最小资源粒度权限。
自定义权限策略脚本如下:
脚本中的
Project名称
和Logstore名称
请根据实际情况替换。为RAM配置更多权限请参见日志服务RAM访问控制权限配置和授权信息。
{ "Version": "1", "Statement": [ { "Action": [ "log:DeleteLogStoreLogs", "log:GetDeleteLogStoreLogsTask", "log:ListDeleteLogStoreLogsTasks", ], "Resource": [ "acs:log:*:*:project/Project名称/logstore/LogStore名称" ] "Effect": "Allow" } ] }
控制台操作
登录日志服务控制台,在Project列表,单击打开目标Project。
在
页签中,单击目标Logstore。在Logstore的查询和分析页面,输入查询语句,然后选择时间范围。在原始日志页签中,单击
。在日志软删除对话框中,请仔细阅读对话框信息后,单击确认删除日志。删除完成后,日志不可见。
删除完成后,在原始日志页签中,单击
。在软删除任务列表中,可以查看历史删除任务。
SDK操作
Python
使用
pip show aliyun-log-python-sdk
检查aliyun-log-python-sdk
版本。aliyun-log-python-sdk
版本要求不能低于0.9.28,如果低于该版本,使用pip install -U aliyun-log-python-sdk
升级SDK版本。提交软删除任务。.
DeleteLogsRequest
参数说明如下:参数说明
示例
# 导入必要的库 import time from aliyun.log import LogClient, DeleteLogsRequest, DeleteLogsResponse def execute_soft_delete(client: LogClient, project: str, logstore: str, from_time: int, to_time: int, query: str) -> str: """ 执行一个异步的软删除任务。 :param client: 已初始化的SLS客户端实例。 :param project: 您的SLS项目名称。 :param logstore: 您的Logstore名称。 :param from_time: 删除范围的开始时间戳(秒)。 :param to_time: 删除范围的结束时间戳(秒)。 :param query: 用来筛选要删除日志的纯索引查询语句(不支持sql、scan、短语查询语句)。 :return: 异步删除任务的Task ID,用于后续查询任务状态。 """ print(f"准备在项目'{project}'的Logstore'{logstore}'中执行软删除...") print(f"时间范围: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(from_time))} -> {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(to_time))}") print(f"删除条件(Query): {query}") # 1. 构造删除请求对象 request = DeleteLogsRequest(project, logstore, from_time, to_time, query=query) # 2. 发起异步删除请求 try: res: DeleteLogsResponse = client.delete_logs(request) # 3. 打印请求信息和返回的任务ID print("\n--- 请求回执 ---") res.log_print() task_id = res.get_taskid() print(f"✅ 软删除任务已成功提交!Request ID: {res.get_request_id()}, Task ID: {task_id}") return task_id except Exception as e: print(f"❌ 软删除任务提交失败: {e}") return "" # --- 如何调用 --- # client = LogClient(...) # 此处省略客户端初始化过程 # project_name = "my-prod-app" # logstore_name = "nginx-access-log" # # 场景:删除过去24小时内,所有状态码为500且来自特定IP的访问日志 # start_time = int(time.time()) - 86400 # end_time = int(time.time()) # delete_query = "__source__: 123.123.XXX.XXX and status: 500" # task_id = execute_soft_delete(client, project_name, logstore_name, start_time, end_time, delete_query) # if task_id: # print(f"\n请使用Task ID '{task_id}' 查询任务执行状态。")
project
String
(必选)Project名称:项目(Project)是日志服务的资源管理单元,是进行多用户隔离与访问控制的主要边界。更多信息,请参见管理Project。
logstore
String
(必选)Logstore名称:Logstore是日志服务中日志数据的采集、存储和查询单元。更多信息,请参见管理Logstore。
fromTime
int
(必选)查询开始的时间点,使用Unix时间戳格式。
toTime
int
(必选)查询结束的时间点,使用Unix时间戳格式。
query
String
(必选)查询语句或分析语句。更多信息,请参见查询与分析概述。
请谨慎设置
"*"
或"level: ERROR"
,这可能会导致大范围的数据被意外删除。轮询任务状态。
该段代码会持续检查任务进度,直到其完成或超时,这是确保删除操作成功执行的关键步骤。
delete_logs
是一个异步API。调用成功仅代表任务被接受,您需要使用返回的task_id
来轮询任务的最终状态。from aliyun.log import GetDeleteLogsStatusRequest, GetDeleteLogsStatusResponse def wait_for_task_completion(client: LogClient, project: str, logstore: str, task_id: str, timeout_seconds: int = 300): """通过轮询等待软删除任务完成。""" start_time = time.time() while time.time() - start_time < timeout_seconds: request = GetDeleteLogsStatusRequest(project, logstore, task_id) try: res: GetDeleteLogsStatusResponse = client.get_delete_logs_status(request) progress = res.get_process() print(f"查询任务 '{task_id}' 状态...进度: {progress:.2f}%") if progress >= 100.0: print("✅ 任务已完成!") return True time.sleep(2) # 避免过于频繁的轮询 except Exception as e: print(f"❌ 查询任务状态失败: {e}") return False print(f"⌛️ 任务超时(超过 {timeout_seconds} 秒),请稍后重试或检查任务状态。") return False # --- 如何调用 --- # task_id = execute_soft_delete(...) # if task_id: # wait_for_task_completion(client, project_name, logstore_name, task_id)
审计与管理(查看历史任务)。
该功能对于安全审计和运维管理来说至关重要,它提供了所有删除操作的完整历史记录。
from aliyun.log import ListDeleteLogsTasksRequest def list_all_delete_tasks(client: LogClient, project: str, logstore: str): """列出指定Logstore下的所有软删除任务历史。""" print(f"\n正在列出项目'{project}'的Logstore'{logstore}'下的软删除任务历史...") request = ListDeleteLogsTasksRequest(project=project, logstore=logstore) try: res = client.list_delete_logs_tasks(request) print("✅ 任务列表获取成功!") res.log_print() # log_print()会以易读的格式打印所有任务详情 except Exception as e: print(f"❌ 获取任务列表失败: {e}")