PythonとAzure Log Analyticsクエリの実行

Pythonを使用してAzure MonitorのLog Analyticsにクエリを実行し、ログを取得する方法について説明します。

必要なライブラリ

以下のPythonライブラリが必要です。

azure-loganalytics
azure-monitor-query
azure-identity
pandas

これらのライブラリはpipを使用してインストールできます。

サンプルコード

以下に、Azure Log Analyticsからログを取得するためのPythonコードのサンプルを示します。

import pandas as pd
from datetime import datetime, timezone
from azure.monitor.query import LogsQueryClient, LogsQueryStatus
from azure.identity import DefaultAzureCredential
from azure.identity import AzureCliCredential
from azure.core.exceptions import HttpResponseError

# Azure CLI で az login した認証情報を使う
credential = AzureCliCredential()

client = LogsQueryClient(credential)

# 発行するクエリを指定
query = "AppServiceConsoleLogs | where _ResourceId contains  'xxxxxxx'"

# ワークスペースID を指定
LOG_WORKSPACE_ID = "xxxx-xx-xx-xxx-x"

# 取得期間を指定
start_time = datetime(2023, 4, 1, tzinfo=timezone.utc)
end_time = datetime(2023, 10, 1, tzinfo=timezone.utc)

try:
    response = client.query_workspace(
        workspace_id=LOG_WORKSPACE_ID,
        query=query,
        timespan=(start_time, end_time)
    )

    if response.status == LogsQueryStatus.PARTIAL:
        error = response.partial_error
        data = response.partial_data
        print(error)
    elif response.status == LogsQueryStatus.SUCCESS:
        data = response.tables
        for table in data:
            df = pd.DataFrame(data=table.rows, columns=table.columns)
            # 取得内容をとりあえず出力
            print(df)
except HttpResponseError as err:
    print("something fatal happened")
    print(err)

このコードは、指定したクエリを使用してAzure Log Analyticsからログを取得し、取得したログをpandasのDataFrameに変換して出力します。

注意点

大量のデータを取得した場合、取得できるかどうか、また処理速度はどれくらいかなど、未確認の点や今後の課題があります。

以上、Pythonを使用してAzure Log Analyticsからログを取得する方法について説明しました。この情報が皆さんの開発に役立つことを願っています。