Simple Log Service:Simple Log Service の論理削除

Python

1. 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 をアップグレードします。

2. 論理削除タスクを送信します。

   次の表に、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、スキャン、およびフレーズクエリはサポートされていません)。 :return: 後でタスクのステータスをクエリするために使用される非同期削除タスクのタスク 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}") # 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"✅ 論理削除タスクが正常に送信されました! リクエスト ID: {res.get_request_id()}, タスク 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" # # シナリオ: 特定の IP アドレスから過去 24 時間のステータスコード 500 のアクセスログを削除します # 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タスク ID '{task_id}' を使用してタスクの実行ステータスをクエリしてください。")
   project String (必須) プロジェクトの名前。Simple Log Service のプロジェクトは、異なるユーザーのリソースを分離し、特定のリソースへのアクセスを制御するために使用されます。詳細については、「プロジェクトの管理」をご参照ください。
   logstore String (必須) Logstore の名前。Simple Log Service の Logstore は、ログの収集、保存、クエリに使用されます。詳細については、「Logstore の管理」をご参照ください。
   fromTime int (必須) クエリする時間範囲の開始時刻。UNIX タイムスタンプとして時間を指定します。
   toTime int (必須) クエリする時間範囲の終了時刻。UNIX タイムスタンプとして時間を指定します。
   query String (必須) 検索文または分析文。詳細については、「クエリと分析の概要」をご参照ください。 クエリを "*" または "level: ERROR" に設定する場合は注意してください。これにより、大量のデータが予期せず削除される可能性があります。

3. タスクステータスをポーリングします。

   このコードブロックは、タスクが完了するかタイムアウトするまで、タスクの進行状況を継続的にチェックします。これは、削除操作が正常に実行されたことを確認するための重要なステップです。

   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)
   

4. 削除タスクの監査と管理 (履歴タスクの表示)。

   この機能は、すべての削除操作の完全な履歴を提供するため、セキュリティ監査と運用管理にとって重要です。

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