Azure Monitor 摄取 Java
Azure Monitor Ingestion SDK for Java
用于通过数据收集规则调用 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>
前置条件
- 数据收集终结点 (DCE)
- 数据收集规则 (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();
核心概念
| 概念 | 描述 |
|---------|-------------|
| 数据收集终结点 (DCE) | 您所在区域的摄取终结点 URL |
| 数据收集规则 (DCR) | 定义数据的转换以及路由到目标表的方式 |
| 流名称 (Stream Name) | 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(); // 大规模集合
LogsUploadOptions options = new LogsUploadOptions()
.setMaxConcurrency(3);
client.upload("<data-collection-rule-id>", "<str
eam-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());
// 选项 1：记录日志并继续
// 选项 2：抛出异常以中止剩余上传
// 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;
}
// JSON 序列化需要 Getter 方法
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. 处理部分失败 — 使用错误消费者 (error consumer) 来记录失败的条目
4. 匹配 DCR 架构 — 日志条目字段必须符合 DCR 转换的预期
5. 包含 TimeGenerated — 大多数表都需要时间戳字段
6. 复用客户端 — 创建一次，在整个应用程序中复用
7. 高吞吐量使用异步 — 响应式模式请使用 LogsIngestionAsyncClient
查询上传的日志
使用 azure-monitor-query 查询已摄入的日志：
// 关于 LogsQueryClient 的用法请参阅 azure-monitor-query 技能
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 |
使用场景
本技能适用于执行概览中所描述的工作流或操作。局限性
- 仅在任务明确符合上述范围时使用本技能。
- 不要将输出结果视为针对特定环境的验证、测试或专家评审的替代方案。
- 如果缺少必要的输入、权限、安全边界或成功标准，请停止操作并寻求澄清。
