Simple Log Service:Exclusão lógica no Simple Log Service

Python

1. Use pip show aliyun-log-python-sdk para verificar a versão do aliyun-log-python-sdk.

   A versão do aliyun-log-python-sdk deve ser 0.9.28 ou superior. Se sua versão for anterior, use o comando pip install -U aliyun-log-python-sdk para atualizar o SDK.

2. Envie uma tarefa de exclusão lógica.

   DeleteLogsRequest possui os seguintes parâmetros:

   Parâmetro	Exemplo # 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 (Obrigatório) Nome do projeto. No Simple Log Service, o projeto isola os recursos de diferentes usuários e controla o acesso a recursos específicos. Consulte Gerenciar projetos.
   LogStore String (Obrigatório) Nome do logstore. No Simple Log Service, o logstore serve para coletar, armazenar e consultar logs. Consulte Gerenciar Logstores.
   fromTime int Obrigatório Início do intervalo de tempo da consulta, especificado como um timestamp Unix.
   toTime int (Obrigatório) Fim do intervalo de tempo da consulta, especificado como um timestamp Unix.
   query String (Obrigatório) Instrução de pesquisa ou análise. Para mais informações, consulte Visão geral de consulta e análise. Tenha cuidado ao definir "*" ou "level: ERROR". Essas configurações podem causar a exclusão acidental de uma grande quantidade de dados.

3. Verifique o status da tarefa periodicamente.

   Este bloco de código verifica continuamente o progresso da tarefa até sua conclusão ou timeout. Trata-se de uma etapa fundamental para garantir a execução bem-sucedida da operação de exclusão.

   delete_logs é uma API assíncrona. Uma chamada bem-sucedida indica apenas que a tarefa foi aceita. Use o task_id retornado para consultar o status final da tarefa.

   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. Audite e gerencie tarefas (visualize tarefas históricas).

   Esse recurso é crucial para auditorias de segurança e gerenciamento operacional, pois fornece um histórico completo de todas as operações de exclusão.

   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}")