Simple Log Service:Soft deletion in Simple Log Service

Python

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

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

2. Submit a soft deletion task.

   DeleteLogsRequest has the following parameters:

   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 Simple Log Service client instance. :param project: Your Simple Log Service 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 Logstores.
   fromTime int Required The start of the query time range, specified as a Unix timestamp.
   toTime int (Required) The end of the query time range, specified as a Unix timestamp.
   query String (Required) The search or analytic statement. For more information, see Query and analysis overview. Use caution when setting "*" or "level: ERROR". These settings may cause the accidental deletion of a large amount of data.

3. Poll the task status.

   This code block continuously checks the task progress until it 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 only indicates 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 tasks (view historical tasks).

   This feature is crucial for security audits and operations management as 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}")