Azure Monitor 数据摄取 Python SDK

azure-monitor-ingestion-py
分类编程
作者Agentic Awesome Skills 社区
许可MIT
评分4.50/5
使用12.2K

Azure Monitor Ingestion Python SDK

使用 Logs Ingestion API 将自定义日志发送到 Azure Monitor Log Analytics 工作区。

安装

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

环境变量

bash
# 数据收集终结点 (DCE)
AZURE_DCE_ENDPOINT=https://<dce-name>.<region>.ingest.monitor.azure.com

数据收集规则 (DCR) 不可变 ID

AZURE_DCR_RULE_ID=dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

DCR 中的流名称

AZURE_DCR_STREAM_NAME=Custom-MyTable_CL

前置条件

在使用此 SDK 之前，您需要：

1. Log Analytics 工作区 — 日志的目标存储地
2. 数据收集终结点 (DCE) — 摄入终结点
3. 数据收集规则 (DCR) — 定义架构和目的地
4. 自定义表 — 位于 Log Analytics 中（通过 DCR 或手动创建）

身份验证

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

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

上传自定义日志

python
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 文件上传

python
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)

自定义错误处理

使用回调函数处理部分失败的情况：

python
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
)


重试失败的日志

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

忽略错误

python
def ignore_errors(error):
    pass  # 静默忽略上传失败

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

异步客户端

python
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())

主权云 (Sovereign Clouds)

python
from azure.identity import Azure
AuthorityHosts, 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 压缩每个分块

  • 并行上传分块

对于大型日志集，无需手动进行分批处理。

客户端类型

| 客户端 | 用途 |
|--------|---------|
| LogsIngestionClient | 用于上传日志的同步客户端 |
| LogsIngestionClient (aio) | 用于上传日志的异步客户端 |

核心概念

| 概念 | 描述 |
|---------|-------------|
| DCE | 数据收集终结点 (Data Collection Endpoint) —— 摄入 URL |
| DCR | 数据收集规则 (Data Collection Rule) —— 定义架构、转换和目的地 |
| Stream | DCR 中的命名数据流 |
| Custom Table | 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. 使用上下文管理器 —— 确保客户端正确清理

适用场景

此技能适用于执行概览中描述的工作流或操作。

局限性

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