azure-monitor-ingestion-py

Azure Monitor Python 引入 SDK

使用 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. Data Collection Endpoint（DCE） — 引入端点
3. Data Collection Rule（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()
)

请使用已授权的端点、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	Data Collection Endpoint — 引入 URL
DCR	Data Collection Rule — 定义架构、转换和目标位置
流	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. 使用上下文管理器 — 确保正确清理客户端

适用场景

当任务涉及执行概述中描述的工作流或操作时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为针对具体环境进行验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请暂停并请求澄清。