azure-monitor-ingestion-py

用途

Azure Monitor Ingestion SDK for Python 用于通过 Logs Ingestion API 将自定义日志发送到 Azure Monitor Log Analytics 工作区。它适合需要把应用或系统产生的自定义日志写入 Log Analytics 的场景。

主要能力

提供同步客户端 LogsIngestionClient 与异步客户端（azure.monitor.ingestion.aio 下的 LogsIngestionClient）。
支持从内存中的日志列表上传，也支持从 JSON 文件读取后上传。
自动批处理：SDK 会将日志拆分为不超过 1MB 的分块，对每个分块进行 gzip 压缩，并并行上传分块，无需手动分批。
支持自定义错误处理：通过 on_error 回调处理部分失败，可收集失败日志并重试，也可静默忽略错误。
支持主权云（如 Azure Government），可指定 authority 与 credential_scopes。

使用方式

安装：

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

前置条件：使用前需要 Log Analytics 工作区（日志目标）、数据收集终结点（DCE，摄取终结点）、数据收集规则（DCR，定义架构与目标）以及 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()
)

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)

忽略错误：

def ignore_errors(error):
    pass  # Silently ignore upload failures

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

异步客户端：

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

关键概念

DCE：数据收集终结点，即摄取 URL。
DCR：数据收集规则，定义架构、转换与目标。
Stream：DCR 中的命名数据流。
Custom Table：Log Analytics 中的目标表，名称以 _CL 结尾。

DCR 流名称格式：

Custom-<TableName>_CL：用于自定义表。
Microsoft-<TableName>：用于内置表。

适用场景

适用于需要将自定义日志写入 Azure Monitor Log Analytics 工作区的任务，例如应用日志采集、系统事件上报等。

最佳实践

使用 DefaultAzureCredential 进行认证。
优雅处理错误，使用 on_error 回调处理部分失败。
包含 TimeGenerated，这是所有日志的必填字段。
匹配 DCR 架构，日志字段必须与 DCR 列定义一致。
高吞吐场景使用异步客户端。
批量上传，SDK 会处理批处理，但应发送合理大小的分块。
监控摄取，检查 Log Analytics 中的摄取状态。
使用上下文管理器，确保客户端正确清理。

限制

仅当任务明确匹配上述范围时使用。
不应将输出视为环境特定验证、测试或专家审查的替代。
如果缺少必需输入、权限、安全边界或成功标准，应停止并请求澄清。