azure-monitor-ingestion-py

Azure Monitor Ingestion SDK for Python

Logs Ingestion API を使用して、カスタムログを Azure Monitor Log Analytics ワークスペースに送信します。

インストール

pip install azure-monitor-ingestion
pip install azure-identity

環境変数

# Data Collection Endpoint (DCE)
AZURE_DCE_ENDPOINT=https://<dce-name>.<region>.ingest.monitor.azure.com

# Data Collection Rule (DCR) immutable ID
AZURE_DCR_RULE_ID=dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

# Stream name from DCR
AZURE_DCR_STREAM_NAME=Custom-MyTable_CL

前提条件

この SDK を使用するには、次のものが必要です。

1. Log Analytics ワークスペース — ログの送信先です
2. データ収集エンドポイント（DCE） — 取り込みエンドポイントです
3. データ収集ルール（DCR） — スキーマと送信先を定義します
4. カスタムテーブル — Log Analytics 内に作成します（DCR 経由または手動で作成）

認証

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

client = LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=DefaultAzureCredential()
)

権限の範囲を限定した ID を使用し、承認済みのエンドポイント／DCR／ストリームとスキーマを指定してください。DCE と DCR のどちらの取り込みエンドポイントを使用するかは、セットアップによって異なります。以下の 2024 年のタイムスタンプは過去のサンプルデータであり、現在のライブ取り込みを検証するものではありません。検証には承認済みの現在時刻に基づく合成タイムスタンプを使用し、アップロード前に PII やシークレットをマスキングして、保持期間と料金を確認してください。失敗時を含め、クライアントと資格情報の両方をクローズしてください。このガイドの検証だけを目的にログをアップロードしてはいけません。

カスタムログをアップロードする

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

client = LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=DefaultAzureCredential()
)

rule_id = os.environ["AZURE_DCR_RULE_ID"]
stream_name = os.environ["AZURE_DCR_STREAM_NAME"]

logs = [
    {"TimeGenerated": "2024-01-15T10:00:00Z", "Computer": "server1", "Message": "Application started"},
    {"TimeGenerated": "2024-01-15T10:01:00Z", "Computer": "server1", "Message": "Processing request"},
    {"TimeGenerated": "2024-01-15T10:02:00Z", "Computer": "server2", "Message": "Connection established"}
]

client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

JSON ファイルからアップロードする

import json

with open("logs.json", "r") as f:
    logs = json.load(f)

client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

カスタムエラー処理

コールバックを使用して部分的な失敗を処理します。

failed_logs = []

def on_error(error):
    print(f"Upload failed: {error.error}")
    failed_logs.extend(error.failed_logs)

client.upload(
    rule_id=rule_id,
    stream_name=stream_name,
    logs=logs,
    on_error=on_error
)

# Retry failed logs
if failed_logs:
    print(f"Retrying {len(failed_logs)} failed logs...")
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=failed_logs)

不完全なアップロード

on_error コールバックは、既定の例外を抑制することがあります。範囲を限定して保護された失敗サマリーを保存し、アップロードを部分成功または失敗として報告してください。失敗したログを黙って破棄したり、バッチ全体を成功と表示したりしてはいけません。分類された一時的な失敗に限り、バックオフの上限を設定して再試行し、結果が不確定な場合の照合と重複取り込みの影響に対処してください。

非同期クライアント

import asyncio
from azure.monitor.ingestion.aio import LogsIngestionClient
from azure.identity.aio import DefaultAzureCredential

async def upload_logs():
    async with LogsIngestionClient(
        endpoint=endpoint,
        credential=DefaultAzureCredential()
    ) as client:
        await client.upload(
            rule_id=rule_id,
            stream_name=stream_name,
            logs=logs
        )

asyncio.run(upload_logs())

ソブリンクラウド

from azure.identity import AzureAuthorityHosts, DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient

# Azure Government
credential = DefaultAzureCredential(authority=AzureAuthorityHosts.AZURE_GOVERNMENT)
client = LogsIngestionClient(
    endpoint="https://example.ingest.monitor.azure.us",
    credential=credential,
    credential_scopes=["https://monitor.azure.us/.default"]
)

バッチ処理の動作

SDK は自動的に次の処理を行います。

• ログを 1MB 以下のチャンクに分割します
• 各チャンクを gzip で圧縮します
• チャンクを並行してアップロードします

SDK によるチャンク分割では、アプリケーションの入力メモリ使用量や取り込みの総費用は制限されません。入力のバイト数／行数に上限を設定し、インストール済みバージョンのバッチ処理と並行処理の動作を確認して、部分成功を適切に処理してください。

クライアントの種類

クライアント	用途
LogsIngestionClient	ログをアップロードする同期クライアント
LogsIngestionClient (aio)	ログをアップロードする非同期クライアント

主な概念

概念	説明
DCE	データ収集エンドポイント — 取り込み URL
DCR	データ収集ルール — スキーマ、変換、送信先を定義します
ストリーム	DCR 内の名前付きデータフロー
カスタムテーブル	Log Analytics の送信先テーブル（末尾が _CL）

DCR ストリーム名の形式

ストリーム名は次の形式に従います。

• Custom-<TableName>_CL — カスタムテーブル用
• Microsoft-<TableName> — 組み込みテーブル用

ベストプラクティス

1. 認証には DefaultAzureCredential を使用します
2. エラーを適切に処理する — 部分的な失敗には on_error コールバックを使用します
3. TimeGenerated を含める — すべてのログで必須のフィールドです
4. DCR スキーマに一致させる — ログのフィールドは DCR の列定義と一致する必要があります
5. スループットが高い場合は 非同期クライアント を使用します
6. バッチでアップロードする — SDK がバッチ処理を行いますが、適切なサイズのチャンクを送信してください
7. 取り込みを監視する — Log Analytics で取り込み状況を確認します
8. コンテキストマネージャーを使用する — クライアントが適切にクリーンアップされます

使用する場合

このスキルは、概要に記載されているワークフローまたはアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の説明の対象範囲に明確に該当する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認を求めてください。