Zenn
📑

Azure Log Analytics / Logs Ingestion API に Python でログを送る例

2025/03/17に公開
Azure
tech

ここまで Azure Functions について、少しずつ調べつつまとめてきましたが、本命は "Azure Functions を利用して Log Analytics にログを取り込みたい" というのものだったので、3回目としてまとめたいと思います。

実現したい構成

  1. Blob へのファイルアップロードをトリガーに Azure Functions でログを処理したい。
  2. 結果のログを Log Analytics (Sentinel) に取り込みたい。

"1" については前回・前々回の記事を見て貰えればイメージできると思うので、今回は "3" の部分だけ軽く試してみたいと思います。

Log Analytics への接続は、本当はプライベートエンドポイント経由にしたかったけど、ちょっと環境を整えてる時間がないので、一旦 Azure Functions -> Log Analytics はパブリックで飛ばします。

データ収集エンドポイントの作成

https://learn.microsoft.com/ja-jp/azure/azure-monitor/logs/tutorial-logs-ingestion-portal

Azure Monitor / Log Analytics にログを取り込むには "Logs ingestion API" を利用します。 この API を有効にするためには幾つか手順があるので、まずはそれを進めていきます。

Log Analytics ワークスペースの作成

最初にログを格納するための Log Analytics ワークスペースを作成します。
Azure Portal から Log Analytics を選択して、任意の名前で作成するだけなので詳細は公式ドキュメントを見てください (リージョンは Japan East を選択しました)

https://learn.microsoft.com/ja-jp/azure/azure-monitor/logs/quick-create-workspace?tabs=azure-portal

Microsoft Entra アプリケーションの作成

引き続き下記のチュートリアルを参考に、リソースを作成していきます。
https://learn.microsoft.com/ja-jp/azure/azure-monitor/logs/tutorial-logs-ingestion-portal

  1. Azure Portal -> Microsoft Entra ID -> アプリの登録 -> 新規登録 から、新しいアプリケーションを登録する。
  2. アプリ作成後、そのリソースを選択して下記２つの値をメモする 📝
    • アプリケーション (クライアント) ID
    • ディレクトリ (テナント) ID
  3. アプリケーションが作成されたら、次にシークレットを作成する。
    作成したアプリを選択 -> 証明書とシークレット -> 新しいクライアントシークレット で作成
    • 作成された クライアントシークレットの値 をメモする 📝

データ収集エンドポイント(DCE)の作成

次にデータ収集エンドポイントを作成します・・・が、公式ドキュメントには下記の記述があります。
今回は Azure Portal から作業する例で書いているので作成しますが Resource Manager (ARMテンプレート) を利用して作成する場合は不要に見えるので適宜選択・対応しましょう。

DCR のエンドポイントを使用できるため、ほとんどの場合、DCE は不要になりました。 ただし、Azure portal では、この変更を反映するための更新がまだ行われていないため、カスタム ログを作成するときは、引き続き DCE が必要です。 他のメソッドを使用してカスタム テーブルと DCR を作成する場合は、DCE ではなく DCR エンドポイントを使用できます。

  1. Azure Portal -> 監視 -> データ収集エンドポイント -> 作成
  2. 作成時には リージョン は Log Analytics と合わせます (例: Japan East)
  3. 作成後 ログインジェスト の URL をメモしておきます 📝

Log Analytics ワークスペースへの新規テーブル作成 (DCR作成)

  1. 事前に作成していた Log Analytics ワークスペース から テーブルを選択 -> 作成 -> 新しいカスタム ログ (DCR ベース) を選択
  2. テーブル名は任意、データ収集ルールは Azure サブスクと、リソースグループを選択、名前は任意
  3. データ収集エンドポイントは、前段で作成していたものを選択し作成する。

サンプルデータのアップロード

次に作成しようとしているテーブルのサンプルとなる JSON ファイルをアップロードする。
今回は下記ページにあるコードも参考にしながら進めたいので、下記コードの内容を JSON ファイル (sample.json) として保存しアップロードする。

https://learn.microsoft.com/ja-jp/azure/azure-monitor/logs/tutorial-logs-ingestion-code?tabs=python

sample.json
[
{
    "Time": "$currentTime",
    "Computer": "Computer1",
    "AdditionalContext": {
        "InstanceName": "user1",
        "TimeZone": "Pacific Time",
        "Level": 4,
        "CounterName": "AppMetric1",
        "CounterValue": 15.3    
    }
},
{
    "Time": "$currentTime",
    "Computer": "Computer2",
    "AdditionalContext": {
        "InstanceName": "user2",
        "TimeZone": "Central Time",
        "Level": 3,
        "CounterName": "AppMetric1",
        "CounterValue": 23.5     
    }
}
]

実際にアップロードをすると、下図のようにテーブルが構成される事が表示されるので、作成を進める。

DCR から API に必要な ID を確認

https://learn.microsoft.com/ja-jp/azure/azure-monitor/logs/tutorial-logs-ingestion-portal

前段で DCR を作成したので、その中にある "immutableId" を収集する。

  1. Azure Portal -> 監視 -> データ収集ルール -> 前段で作成した DCR 名を選択
  2. 概要 から [JSON ビュー] を選択
  3. JSON にある "dcr-" から始まる "immutableId" と "streamDeclarations" にある値 (Custom-[TableName]) をメモする 📝

DCR にアプリケーションのアクセス許可を割り当て

作成していた DCR リソースに対しアプリケーションからログを投げるためのアクセス許可を与える。

  1. DCRリソースを選択 -> アクセス制御 (IAM) -> ロールの割り当ての追加を選択
  2. 事前に作成していた Entra ID アプリケーションに対して "監視メトリック発行者" のロールと割り当てる。

ログの送信テスト

前段では sample.json を用いて、Log Analytics にテーブルを作成しましたが、今回は Python を用いて Log Analytics に取り込むところまで実行したいと思います。

下記ページにあるサンプルコードを元に、PowerShell 経由ではなく Python だけで実行できるコードに書き直して試してみます。

https://learn.microsoft.com/ja-jp/azure/azure-monitor/logs/tutorial-logs-ingestion-code?tabs=powershell

サンプルコード

sample.py
import requests
import urllib.parse
import json
from datetime import datetime

### Step 0: Set variables required for the rest of the script.
# 情報は適宜実際の値に置き換えてください
tenantId = "00000000-0000-0000-00000000000000000"  # テナントID
appId = "000000000-0000-0000-00000000000000000"    # Entra ID アプリケーションID
appSecret = "0000000000000000000000000000000000000000"  # Entra ID アプリケーションIDシークレット

# Information needed to send data to the DCR endpoint
endpoint_uri = "https://my-url.monitor.azure.com"  # DCE作成時にメモした URL
dcrImmutableId = "dcr-00000000000000000000000000000000"  # DCR で確認した immutableId
streamName = "Custom-MyTableRawData"  # DCR で確認した streamName (Custom-[送信先テーブル名])

### Step 1: Obtain a bearer token used later to authenticate against the DCR.
scope = urllib.parse.quote("https://monitor.azure.com//.default", safe="")
body = {
    "client_id": appId,
    "scope": "https://monitor.azure.com//.default",
    "client_secret": appSecret,
    "grant_type": "client_credentials"
}
uri = f"https://login.microsoftonline.com/{tenantId}/oauth2/v2.0/token"

try:
    response = requests.post(uri, data=body, headers={"Content-Type": "application/x-www-form-urlencoded"})
    response.raise_for_status()  # HTTPエラー時に例外を投げる
    bearerToken = response.json()["access_token"]
    print("Bearer token acquired successfully.")
except requests.exceptions.RequestException as e:
    print("Error acquiring bearer token:", e)
    exit(1)

### Step 2: Create some sample data.
currentTime = datetime.utcnow().isoformat() + "Z"  # ISO 8601 format (UTC)
staticData = [
    {
        "Time": currentTime,
        "Computer": "Computer1",
        "AdditionalContext": {
            "InstanceName": "user1",
            "TimeZone": "Pacific Time",
            "Level": 4,
            "CounterName": "AppMetric1",
            "CounterValue": 15.3
        }
    },
    {
        "Time": currentTime,
        "Computer": "Computer2",
        "AdditionalContext": {
            "InstanceName": "user2",
            "TimeZone": "Central Time",
            "Level": 3,
            "CounterName": "AppMetric1",
            "CounterValue": 23.5
        }
    }
]

### Step 3: Send the data to the Log Analytics workspace.
headers = {
    "Authorization": f"Bearer {bearerToken}",
    "Content-Type": "application/json"
}
uri = f"{endpoint_uri}/dataCollectionRules/{dcrImmutableId}/streams/{streamName}?api-version=2023-01-01"

try:
    uploadResponse = requests.post(uri, json=staticData, headers=headers)
    # ステータスコードを表示
    print("Status Code:", uploadResponse.status_code)
    
    # 成功時（200, 202, 204など）の場合はボディが空またはJSONでないことがある
    # まずテキストを表示して確認
    print("Response Text:", uploadResponse.text)

    # JSONとしてパースを試みる
    try:
        uploadResponseJson = uploadResponse.json()
        print("Upload Response (JSON):", uploadResponseJson)
    except json.JSONDecodeError:
        print("Response is not valid JSON or is empty. Skipping JSON parse.")

    uploadResponse.raise_for_status()

    # 成功メッセージ
    print("Data uploaded successfully.")

except requests.exceptions.RequestException as e:
    print("Error uploading data:", e)
    exit(1)

実行

上記スクリプトを実行すると下記のような出力が出てくる。 応答コード 204 で戻り値がないが、もし認証やなにかしらのエラーであれば 4xx でコケるのでエラー内容に応じて適宜修正する。

$ python3 hogehoge2.py
Bearer token acquired successfully.
Status Code: 204
Response Text:
Response is not valid JSON or is empty. Skipping JSON parse.
Data uploaded successfully.

Log Analytics でのデータ確認

Azure Portal に戻り、実際にログが取り込まれたのか確認をする。

  1. Azure Portal -> Log Analytics ワークスペース -> 作成していたテーブル -> ログ
  2. 問題がなければ、下図のようにログを投げる事が出来た事がわかる。

まとめ

最終的に Blob へのファイルアップロード -> トリガーで Azure Functions を実行 -> Blob のファイルを展開し Log Analytics まで取り込み～まで確認しようと思っていましたが、思っていた以上に環境を作るだけでもめんどくさいので、ちょっと途中で方針変えました（；^ω^）

ただ前回・前々回の内容を踏まえた上で、今回の環境作成まで出来れば後はコードの問題なので、なんとかなると思います。

参考情報・備忘メモ

https://qiita.com/YoshiakiOi/items/466c1660ff2811225624
https://techcommunity.microsoft.com/blog/microsoftsentinelblog/ingesting-akamai-audit-logs-into-microsoft-sentinel-using-azure-function-apps/4395969

Discussion