Simple Log Service:Soft deletion in Simple Log Service

Python

1. Run pip show aliyun-log-python-sdk to check the version of aliyun-log-python-sdk.

   The version of aliyun-log-python-sdk must be 0.9.28 or later. If the version is earlier, run pip install -U aliyun-log-python-sdk to upgrade the SDK.

2. Submit a soft deletion task.

   The following table describes the parameters of DeleteLogsRequest.

   Parameter	Example # Import necessary libraries 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: """ Executes an asynchronous soft deletion task. :param client: An initialized SLS client instance. :param project: Your SLS project name. :param logstore: Your Logstore name. :param from_time: The start timestamp of the deletion range (in seconds). :param to_time: The end timestamp of the deletion range (in seconds). :param query: The index-only search statement used to filter logs for deletion (SQL, scan, and phrase queries are not supported). :return: The task ID of the asynchronous deletion task, used to query the task status later. """ print(f"Preparing to perform a soft deletion in Logstore '{logstore}' of project '{project}'...") print(f"Time range: {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"Deletion condition (Query): {query}") # 1. Construct the deletion request object request = DeleteLogsRequest(project, logstore, from_time, to_time, query=query) # 2. Initiate the asynchronous deletion request try: res: DeleteLogsResponse = client.delete_logs(request) # 3. Print the request information and the returned task ID print("\n--- Request Receipt ---") res.log_print() task_id = res.get_taskid() print(f"✅ Soft deletion task submitted successfully! Request ID: {res.get_request_id()}, Task ID: {task_id}") return task_id except Exception as e: print(f"❌ Failed to submit soft deletion task: {e}") return "" # --- How to call --- # client = LogClient(...) # Client initialization process is omitted here # project_name = "my-prod-app" # logstore_name = "nginx-access-log" # # Scenario: Delete access logs from the last 24 hours with a status code of 500 from a specific IP address # 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"\nPlease use Task ID '{task_id}' to query the task execution status.")
   project String (Required) The name of the project. The project in Simple Log Service is used to isolate the resources of different users and control access to specific resources. See Manage projects.
   logstore String (Required) The name of the logstore. The logstore in Simple Log Service is used to collect, store, and query logs. See Manage a logstore.
   fromTime int (Required) The beginning of the time range to query. Specify the time as a UNIX timestamp.
   toTime int (Required) The end of the time range to query. Specify the time as a UNIX timestamp.
   query String (Required) The search statement or analytic statement. For more information, see Query and analysis overview. Use caution when you set the query to "*" or "level: ERROR". This may cause a large amount of data to be unexpectedly deleted.

3. Poll the task status.

   This code block continuously checks the task progress until the task is complete or times out. This is a key step to ensure that the deletion operation is successfully executed.

   delete_logs is an asynchronous API. A successful call indicates only that the task is accepted. You must use the returned task_id to poll the final status of the task.

   from aliyun.log import GetDeleteLogsStatusRequest, GetDeleteLogsStatusResponse
   
   def wait_for_task_completion(client: LogClient, project: str, logstore: str, task_id: str, timeout_seconds: int = 300):
       """Waits for the soft deletion task to complete by polling."""
       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"Querying status of task '{task_id}'... Progress: {progress:.2f}%")
               if progress >= 100.0:
                   print("✅ Task completed!")
                   return True
               time.sleep(2)  # Avoid polling too frequently
           except Exception as e:
               print(f"❌ Failed to query task status: {e}")
               return False
       print(f"⌛️ Task timed out (exceeded {timeout_seconds} seconds). Please retry later or check the task status.")
       return False
   
   # --- How to call ---
   # task_id = execute_soft_delete(...)
   # if task_id:
   #     wait_for_task_completion(client, project_name, logstore_name, task_id)
   

4. Audit and manage deletion tasks (view historical tasks).

   This feature is crucial for security audits and operations management because it provides a complete history of all deletion operations.

   from aliyun.log import ListDeleteLogsTasksRequest
   
   def list_all_delete_tasks(client: LogClient, project: str, logstore: str):
       """Lists the history of all soft deletion tasks for a specified Logstore."""
       print(f"\nListing the history of soft deletion tasks for Logstore '{logstore}' in project '{project}'...")
       request = ListDeleteLogsTasksRequest(project=project, logstore=logstore)
       try:
           res = client.list_delete_logs_tasks(request)
           print("✅ Task list retrieved successfully!")
           res.log_print() # log_print() prints all task details in a readable format
       except Exception as e:
           print(f"❌ Failed to retrieve task list: {e}")