azure-monitor-ingestion-java

Azure Monitor Ingestion SDK for Java

用于通过 Data Collection Rule，借助 Logs Ingestion API 向 Azure Monitor 发送自定义日志的客户端库。

安装

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-monitor-ingestion</artifactId>
    <version>1.2.11</version>
</dependency>

也可以使用 Azure SDK BOM：

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-sdk-bom</artifactId>
            <version>{bom_version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>com.azure</groupId>
        <artifactId>azure-monitor-ingestion</artifactId>
    </dependency>
</dependencies>

先决条件

• Data Collection Endpoint（DCE）
• Data Collection Rule（DCR）
• Log Analytics 工作区
• 目标表（自定义表或内置表：CommonSecurityLog、SecurityEvents、Syslog、WindowsEvents）

环境变量

DATA_COLLECTION_ENDPOINT=https://<dce-name>.<region>.ingest.monitor.azure.com
DATA_COLLECTION_RULE_ID=dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
STREAM_NAME=Custom-MyTable_CL

创建客户端

同步客户端

import com.azure.identity.DefaultAzureCredential;
import com.azure.identity.DefaultAzureCredentialBuilder;
import com.azure.monitor.ingestion.LogsIngestionClient;
import com.azure.monitor.ingestion.LogsIngestionClientBuilder;

DefaultAzureCredential credential = new DefaultAzureCredentialBuilder().build();

LogsIngestionClient client = new LogsIngestionClientBuilder()
    .endpoint("<data-collection-endpoint>")
    .credential(credential)
    .buildClient();

异步客户端

import com.azure.monitor.ingestion.LogsIngestionAsyncClient;

LogsIngestionAsyncClient asyncClient = new LogsIngestionClientBuilder()
    .endpoint("<data-collection-endpoint>")
    .credential(new DefaultAzureCredentialBuilder().build())
    .buildAsyncClient();

核心概念

概念	描述
Data Collection Endpoint（DCE）	你所在区域的摄取端点 URL
Data Collection Rule（DCR）	定义数据转换方式以及如何将数据路由到表
流名称	DCR 中的目标流（例如 Custom-MyTable_CL）
Log Analytics 工作区	已摄取日志的目标位置

核心操作

上传自定义日志

import java.util.List;
import java.util.ArrayList;

List<Object> logs = new ArrayList<>();
logs.add(new MyLogEntry("2024-01-15T10:30:00Z", "INFO", "Application started"));
logs.add(new MyLogEntry("2024-01-15T10:30:05Z", "DEBUG", "Processing request"));

client.upload("<data-collection-rule-id>", "<stream-name>", logs);
System.out.println("Logs uploaded successfully");

并发上传

对于大型日志集合，请启用并发上传：

import com.azure.monitor.ingestion.models.LogsUploadOptions;
import com.azure.core.util.Context;

List<Object> logs = getLargeLogs(); // Large collection

LogsUploadOptions options = new LogsUploadOptions()
    .setMaxConcurrency(3);

client.upload("<data-collection-rule-id>", "<stream-name>", logs, options, Context.NONE);

上传时处理错误

妥善处理部分上传失败：

LogsUploadOptions options = new LogsUploadOptions()
    .setLogsUploadErrorConsumer(uploadError -> {
        System.err.println("Upload error: " + uploadError.getResponseException().getMessage());
        System.err.println("Failed logs count: " + uploadError.getFailedLogs().size());
        
        // Option 1: Log and continue
        // Option 2: Throw to abort remaining uploads
        // throw uploadError.getResponseException();
    });

client.upload("<data-collection-rule-id>", "<stream-name>", logs, options, Context.NONE);

使用 Reactor 异步上传

import reactor.core.publisher.Mono;

List<Object> logs = getLogs();

asyncClient.upload("<data-collection-rule-id>", "<stream-name>", logs)
    .doOnSuccess(v -> System.out.println("Upload completed"))
    .doOnError(e -> System.err.println("Upload failed: " + e.getMessage()))
    .subscribe();

日志条目模型示例

public class MyLogEntry {
    private String timeGenerated;
    private String level;
    private String message;
    
    public MyLogEntry(String timeGenerated, String level, String message) {
        this.timeGenerated = timeGenerated;
        this.level = level;
        this.message = message;
    }
    
    // Getters required for JSON serialization
    public String getTimeGenerated() { return timeGenerated; }
    public String getLevel() { return level; }
    public String getMessage() { return message; }
}

错误处理

import com.azure.core.exception.HttpResponseException;

try {
    client.upload(ruleId, streamName, logs);
} catch (HttpResponseException e) {
    System.err.println("HTTP Status: " + e.getResponse().getStatusCode());
    System.err.println("Error: " + e.getMessage());
    
    if (e.getResponse().getStatusCode() == 403) {
        System.err.println("Check DCR permissions and managed identity");
    } else if (e.getResponse().getStatusCode() == 404) {
        System.err.println("Verify DCE endpoint and DCR ID");
    }
}

最佳实践

1. 批量处理日志 — 分批上传，而不是一次上传一条
2. 使用并发 — 对于大型上传，设置 maxConcurrency
3. 处理部分失败 — 使用错误回调记录失败的条目
4. 匹配 DCR 架构 — 日志条目字段必须符合 DCR 转换预期
5. 包含 TimeGenerated — 大多数表都需要时间戳字段
6. 重复使用客户端 — 创建一次，并在整个应用程序中重复使用
7. 高吞吐量场景使用异步模式 — 在响应式模式中使用 LogsIngestionAsyncClient

查询已上传的日志

使用 azure-monitor-query 查询已摄取的日志：

// See azure-monitor-query skill for LogsQueryClient usage
String query = "MyTable_CL | where TimeGenerated > ago(1h) | limit 10";

参考链接

资源	URL
Maven 包	https://central.sonatype.com/artifact/com.azure/azure-monitor-ingestion
GitHub	https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/monitor/azure-monitor-ingestion
产品文档	https://learn.microsoft.com/azure/azure-monitor/logs/logs-ingestion-api-overview
DCE 概述	https://learn.microsoft.com/azure/azure-monitor/essentials/data-collection-endpoint-overview
DCR 概述	https://learn.microsoft.com/azure/azure-monitor/essentials/data-collection-rule-overview
故障排除	https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/monitor/azure-monitor-ingestion/TROUBLESHOOTING.md

适用场景

当需要执行概述中描述的工作流或操作时，可使用此技能。

限制

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

请将记录属性名称、时间戳格式和值序列化为与 Data Collection Rule 流架构完全一致的形式；POJO 时间戳或控制台日志并不能证明数据已摄取。等待异步操作完成并处理错误，同时单独检查目标工作区，然后再宣布成功。请使用经过授权的测试流，限制批次量，并避免记录私密记录。