Python SDK を利用した Azure Log Analytics の Log Ingest API (ログインジェスト API) によるログ取り込み

本記事について

前の記事で、Azure Log Analytics の新しいデータ取り込みのための API である Log Ingest API (ログインジェスト API) の使い方を見てみました。

Postman を利用して、Log Ingest API の仕組みを確認することはできましたが。実際に利用するにはプログラミング言語でコーディングし、Azure Functions や Azure Virtual Machines (VM) などのプラットフォームで実行することになるかと思います。

本記事では、Log Ingest API 用の SDK が揃っている Python を使って、Log Ingest API 経由でデータを取り込む方法を見ていきます。

事前準備

前の記事を参考に、データ収集エンドポイント、DCR ベースのカスタムログ（<ログ名>_CL のテーブルとスキーマ)とデータ収集ルール (DCR) を作成してください。

本記事では、簡単な下記 JSON をアップロードして定義されたテーブルを使います。

{
  "TimeGenerated": "2023-01-01 00:00:00",
  "Message": "string"
}

前回の記事の API Call を行った際と同様に、下記情報が必要になるので、メモしておきます。

• データ収集エンドポイントの URI: これはデータ収集エンドポイントのページに書かれています。（ログインジェストと書かれているものです。）（ドメインだけでなく、https://からメモしておきます。）
• DCR の Immutable ID: これは DCR ページで概要を JSON ビューに切り替えると出てきます。
• Stream 名: Custom-<カスタムログのテーブル名 (_CL含む) が Stream 名になります。

また、実行する ID に対して、DCR の監視メトリック発行者の権限を付与してください。Python の DefaultAzureCredential クラスを利用して認証を行う場合、ローカル PC であれば、実行するシェル上で Azure CLI などでログインするアカウントが ID になります。Azure 仮想マシンであれば、マネージド ID を利用することもできます。

DefaultAzureCredential クラスを利用する際に、このクラスが利用する ID についての詳細は、公式ドキュメントをご参照ください。

また、マシン上で ID のログインが行えない場合は、Postman のサンプルで用いた Azure AD アプリケーションを活用することもできます。この場合は、EnvironmentCredential クラスを利用することになります。その場合は、以下３つの環境変数をあらかじめ設定しておき、このクラスから呼び出します。

Linux の場合

export AZURE_TENANT_ID="<テナントID>"
export AZURE_CLIENT_ID="<アプリケーションID>"
export AZURE_CLIENT_SECRET="<サービスプリンシパルのシークレット>"

Python のコードの準備

SDK について

Python 用 Azure Monitor インジェスト クライアント ライブラリというライブラリが用意されていますので、こちらを利用していきます。内部的には Log Ingest API が利用されています。本記事では細かい動作については触れていないので、詳細を確認されたい方は下記をご参照ください。

サンプルのログファイル

下記のようなサンプル JSON ファイルを用意します。これはあとで Python コードから開きます。

{
  "TimeGenerated": "2023-04-10 10:12:00",
  "Message": "Success"
}

実行するコード

今回は、下記のコードをマシン上で実行してみます。アクセス設定を直前に行った場合、Azure AD から発行されるトークンにスコープが入っていない可能性があります。その際は少し時間をおいてから再度お試しください。

import json
from azure.identity import DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient

def main():

    # Azure Monitor の Data Collection Rule の設定
    data_collection_endpoint = "<データ収集エンドポイントのURI>"
    dcr_rule_id = "<DCR の Immutable ID>"
    dcr_stream_name = "<Stream 名>"

    # Azure AD の認証情報 - DefaultAzureCredential クラスを利用する際
    credential = DefaultAzureCredential() # もし特定の ID の利用を避けたい場合は exclude... という引数をTrueで設定します。

    #　Azure AD の認証情報 - EnvironmentCredential クラスを利用する際
    # credential = EnvironmentCredential()

    # ローカルのログファイルを読み込む（フォルダ名とファイル名は変更してください。）
    with open("C:\\Users\\yoshiakioi\\Downloads\\python-log-test1.json", "r") as f:
        logs = json.load(f) 

    # Azure Monitor のログデータを送信するクライアントを作成  
    logs_client = LogsIngestionClient(data_collection_endpoint, credential)

    # Azure Monitor にログデータを送信
    logs_client.upload(rule_id=dcr_rule_id, stream_name=dcr_stream_name, logs=logs)    

if __name__ == "__main__":
    main()

実行結果

Log Analytics 側で検索すると、送信したデータが入っていることが確認できます。

最後に

本記事では、新しくリリースされた Log Ingest API を使ったログデータの取り込みについて、特に Python SDK を使う方法を見ていきました。実際に Log Analytics に Python SDK を使ってカスタムログを取りこむ際には、Azure Functions などを活用することになるかと思いますが、その事前準備として本記事がお役に立てば幸いです。

＊本稿は、個人の見解に基づいた内容であり、所属する会社の公式見解ではありません。また、いかなる保証を与えるものでもありません。正式な情報は、各製品の販売元にご確認ください。