azure-monitor-ingestion-java

Java 用 Azure Monitor Ingestion SDK

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. ログをバッチ処理する — 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 のタイムスタンプやコンソールログだけでは、データが取り込まれた証明にはなりません。エラー処理を行いながら非同期処理の完了を待ち、成功を宣言する前に対象ワークスペースを別途確認してください。認可されたテストストリームを使用し、バッチ量に上限を設け、個人情報を含むレコードをログに記録しないでください。