Azure Monitor Ingestion客户端库，用于Python

Azure Monitor Ingestion客户端库用于通过Azure Monitor使用日志导入API发送自定义日志。

此库允许您从几乎任何来源发送数据到支持的内置表或到您在日志分析工作区中创建的自定义表。您甚至可以扩展内置表的架构以包含自定义列。

资源

• 源代码
• 包（PyPI）
• 包（Conda）
• API参考文档
• 服务文档
• 示例
• 变更日志

入门

先决条件

• Python 3.7或更高版本
• Azure订阅
• Azure Log Analytics工作区
• 数据收集端点
• 数据收集规则

安装软件包

使用pip安装Azure Monitor Ingestion客户端库（Python版）

pip install azure-monitor-ingestion

创建客户端

要将日志上传到Azure Monitor，需要一个认证的客户端。该库包含客户端的同步和异步形式。要认证，创建一个令牌凭证的实例。在创建LogsIngestionClient时使用该实例。以下示例使用了来自azure-identity包的DefaultAzureCredential。

同步客户端

以下示例演示了如何创建用于上传日志的同步客户端

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

endpoint = os.environ['DATA_COLLECTION_ENDPOINT']
credential = DefaultAzureCredential()
logs_client = LogsIngestionClient(endpoint, credential)

异步客户端

客户端API的异步形式可以在以.aio结尾的命名空间中找到。例如

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

endpoint = os.environ['DATA_COLLECTION_ENDPOINT']
credential = DefaultAzureCredential()
logs_client = LogsIngestionClient(endpoint, credential)

配置非公共Azure云的客户端

默认情况下，LogsIngestionClient配置为连接到公共Azure云。要连接到非公共Azure云，需要进行一些额外的配置。必须使用credential_scopes关键字参数提供适当的认证范围。以下示例显示了如何配置客户端以连接到Azure US Government。

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

# Authority can also be set via the AZURE_AUTHORITY_HOST environment variable.
credential = DefaultAzureCredential(authority=AzureAuthorityHosts.AZURE_GOVERNMENT)
logs_client = LogsIngestionClient(endpoint, credential, credential_scopes=["https://monitor.azure.us/.default"])

关键概念

数据收集端点

数据收集端点（DCE）允许您为Azure Monitor唯一配置摄取设置。此文章提供了数据收集端点的概述，包括其内容和结构，以及如何创建和使用它们。

数据收集规则

数据收集规则（DCR）定义了Azure Monitor收集的数据，并指定了这些数据应该如何以及在哪里被发送或存储。必须指定一个DCR来使用REST API调用。单个DCE可以支持多个DCR，因此可以为不同的源和目标表指定不同的DCR。

DCR必须了解输入数据的结构和目标表的结构。如果两者不匹配，它可以使用转换将源数据转换为与目标表匹配。您还可以使用转换来过滤源数据，并执行任何其他计算或转换。

有关更多信息，请参阅Azure Monitor中的数据收集规则，以及有关DCR结构的详细信息，请参阅此文章。有关如何检索DCR ID的信息，请参阅此教程。

日志分析工作区表

自定义日志可以发送到您创建的任何自定义表以及您的日志分析工作区中的某些内置表。在您可以将数据发送到目标表之前，目标表必须存在。以下是目前支持的内置表：

• CommonSecurityLog
• SecurityEvents
• Syslog
• WindowsEvents

日志检索

使用此库上传的日志可以使用Azure Monitor Query客户端库进行查询。

示例

• 上传自定义日志
• 从JSON文件或字符串上传数据
• 使用自定义错误处理上传

上传自定义日志

此示例展示了如何将日志上传到Azure Monitor。

import os

from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient

endpoint = os.environ['DATA_COLLECTION_ENDPOINT']
rule_id = os.environ['LOGS_DCR_RULE_ID']
stream_name = os.environ['LOGS_DCR_STREAM_NAME']

credential = DefaultAzureCredential()
client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True)

body = [
      {
        "Time": "2021-12-08T23:51:14.1104269Z",
        "Computer": "Computer1",
        "AdditionalContext": "context-2"
      },
      {
        "Time": "2021-12-08T23:51:14.1104269Z",
        "Computer": "Computer2",
        "AdditionalContext": "context"
      }
    ]

try:
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=body)
except HttpResponseError as e:
    print(f"Upload failed: {e}")

从JSON文件或字符串上传数据

此示例展示了当数据位于JSON文件或字符串中时的上传。

import json
import os

from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient

endpoint = os.environ["DATA_COLLECTION_ENDPOINT"]
rule_id = os.environ['LOGS_DCR_RULE_ID']
stream_name = os.environ["LOGS_DCR_STREAM_NAME"]

credential = DefaultAzureCredential()
client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True)

# If you have a JSON file containing an array of JSON objects
file_path = "./test-logs.json"
with open(file_path, "r") as f:
    logs = json.load(f)
    try:
        client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)
    except HttpResponseError as e:
        print(f"Upload failed: {e}")

# If you have a JSON string representing an array of JSON objects
string = '[{"Time": "2023-12-08T23:51:14.1104269Z", "Computer": "Computer1", "AdditionalContext": "context-2"}]'
logs = json.loads(string)
try:
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)
except HttpResponseError as e:
    print(f"Upload failed: {e}")

使用自定义错误处理上传

要使用自定义错误处理上传日志，您可以将回调函数传递给upload方法的on_error参数。对于上传过程中发生的每个错误，都会调用回调函数，并期望一个参数对应于一个LogsUploadError对象。该对象包含遇到的错误以及上传失败的日志列表。

# Example 1: Collect all logs that failed to upload.
failed_logs = []
def on_error(error):
    print("Log chunk failed to upload with error: ", error.error)
    failed_logs.extend(error.failed_logs)

# Example 2: Ignore all errors.
def on_error_pass(error):
    pass

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

故障排除

有关诊断各种故障场景的详细信息，请参阅我们的故障排除指南。

下一步操作

要了解更多关于Azure Monitor的信息，请参阅Azure Monitor服务文档。

示例

以下代码示例展示了使用Azure Monitor Ingestion客户端库的常见场景。

日志导入示例

• 上传日志列表 (异步示例)
• 上传带有自定义错误处理的日志列表 (异步示例)
• 上传文件内容 (异步示例)
• 上传pandas DataFrame中的数据 (异步示例)

贡献

本项目欢迎贡献和建议。大多数贡献都需要您同意一个贡献者许可协议（CLA），声明您有权，并且实际上确实授予我们使用您贡献的权利。有关详细信息，请访问cla.microsoft.com。

当您提交拉取请求时，CLA机器人会自动确定您是否需要提供CLA，并适当装饰PR（例如，标签，评论）。只需遵循机器人提供的说明即可。您只需在整个使用我们的CLA的存储库中这样做一次。

本项目已采用Microsoft开源行为准则。有关更多信息，请参阅行为准则常见问题解答或通过opencode@microsoft.com联系以提出任何额外的问题或评论。

发行历史

1.0.4 (2024-06-11)

其他更改

• 将对azure-core的最小依赖性提升到>=1.28.0。
• 在upload方法中对logs参数添加了额外的类型验证，以确保没有传入字符串。(#33976)

1.0.3 (2023-11-07)

其他更改

• 在upload方法中对logs参数添加类型验证。(#32591)

1.0.2 (2023-06-15)

修复了错误

• 修复了阻止将自定义身份验证策略或凭证范围传递给客户端的问题。(#30739)

1.0.1 (2023-04-11)

修复了错误

• 修复了在分块时计算日志条目大小错误的问题。(#29584)

1.0.0 (2023-02-16)

添加了新功能

• 在upload方法中添加了新的on_error参数，允许用户以自己的方式处理错误。
  • 添加了LogsUploadError类来封装有关错误的信息。此类的实例传递给on_error回调。
• 为上传添加了IO支持。现在可以使用logs参数传递IO流。(#28373)

破坏性更改

• 移除了对max_concurrency的支持

其他更改

• 移除了msrest依赖
• 添加了对isodate>=0.6.0的要求（isodate由msrest需要）。
• 添加了对typing-extensions>=4.0.1的要求。

1.0.0b1 (2022-07-15)

功能

• 版本（1.0.0b1）是我们创建用户友好且Pythonic的Azure Monitor Ingestion客户端库努力的第一个预览版。有关更多信息以及其他Azure SDK库的预览版本，请访问https://azure.github.io/azure-sdk/releases/latest/python.html。
• 添加了~azure.monitor.ingestion.LogsIngestionClient以将日志发送到Azure Monitor，同时包括~azure.monitor.ingestion.aio.LogsIngestionClient。