Data Collector API が2026-09-14でサポート終了

おそらくすぐに利用できなくなるわけではなさそうですが、この先もデータを送り続けたい場合は、Logs Ingestion APIに移行する必要があります。

Logs Ingestion APIとは

詳しい説明はドキュメントに譲りますが、Data Collector APIで十分だった人からすれば正直、面倒が増えただけに感じることでしょう

〇 Entra ID認証をサポート（逆を言えば、共有キーではログを送れなくなったので認証処理をしないといけない。時代なので仕方がない）
〇 KQLを使った入力データの変換をサポート（しかし、Data Collector APIを使っているときは送りたいデータの形は決まっているのでそこまで活用しない気も…）
× DCRのリソース定義が必要
× 入力スキーマをちゃんと書かなくてはならない（送る側と受ける側で合わせないといけない）
× Request Sizeで1MB制限（Data Collector API は32MBもあったのでだいぶ小さい）
× DCRを使うためにはLog Analytics Workspaceのテーブルもクラシックから移行しておく必要がある

そして、Logs Ingestion APIを使うとなったら必ず出てくるのが「データ収集ルール（Data Collection Rule=DCR）」です。

なのでDCRを理解しないといけません。

Data Collection Rule(DCR)とは

データを抽出(Extract)・変換(Transform)・読み込み(Load)するための定義を行う、Azure MonitorファミリーのETLソリューションであるリソースの一つです。

Logs Ingestion APIはそのDCRの入力ソース(Extract)の1つといえます。（Extractしてるのは送信側なのでちょっと違うかも）

以降は Data Collector API から移行するために絞って話をします。

DCRの構成図

公式ドキュメントにも図がありますが、概念の説明に留まっていて、実際の実装や設定にどう結びつくのかはイメージしにくかったです。

そのため、実装目線で頭を整理するために書いたのが以下の図です。以降の節では、この図に出てくる各要素（Endpoint、Stream、DataFlow、Destinationなど）を実際の設定に対応させながら説明します。

Endpointが複数あるように見えますが、Private Endpointを使うとかAzure Monitor Agentでログインジェストするといった状況でもない限り、どちらでも同じです。詳しくはドキュメントを参照。DCEを作るとDCRの紐づけなど管理することが増えるので、DCEを使わずにすむなら、DCRのLog Ingestion Endpointを使うほうがよいでしょう。

DCRのImmutable IDは、DCRリソースを一意に識別するためのIDです。Endpointはこの値でDCRリソースを識別します。

Streamは、言わば入力スキーマを明示して名前をつけるイメージです。入力データはこのスキーマに一致するものしか処理できません。後続の処理で変換をするため、そのスキーマを確定させたいからだと考えています。幸い、動的なデータ(dynamic)をサポートしているので、アプリ都合で動的に増える構造ログは扱えるでしょう。

あとはURLで送付先のStreamが決定するので、Streamのスキーマに合わせてJSONデータを送ると、Dataflowで変換されて、Destinationで定義されたLogAnalytics Workspaceへ流れていく、という感じです。

図では、あるStreamから複数のDataFlow、あるDataFlowから複数のDestinationに送る表現を書いていますが、Data Collector APIから引っ越す人は、こんな分岐させることはないでしょう。

DCRのコスト

結論として、Data Collector APIから移行する大半のケースではコストへの影響はほぼありません。

Log AnalyticsのAnalytics/Basicログの場合、課金は以下の式にもとづきます。

[変換によって削除されたデータ量（GB）] - ([受信データサイズ（GB）] / 2)

つまり課金が発生するのは、変換によってデータを削った分が、受信データの半分を超えた場合だけです。Data Collector APIからの移行者は、もともと送信側でデータを絞り込んでいるケースが大半で、DCR側の変換でさらに削るような構成にはなりません。そのため、上の式が実質ゼロになり、コストへの影響はほぼないと考えてよいでしょう。

参考: Basic ログ, Analyticsログ、補助ログのそれぞれの算出例、料金はいずれも$0.145/GBから (料金ページの"ログの処理中"を参照)

Log Analytics WorkspaceのTableのTierにより分かれ、BasicとAnalytics tierの場合はさらに処理した内容でコストが決まります。

注意したいのは、1つのStreamを複数のDataFlowに流すような構成です。この場合は変換するデータ量も増えるため、"削除されたデータ量"が増えたり、単純にIngest自体が増えることがあります。Data Collector APIからの移行するだけならば、1Streamにつき1DataFlowで落ち着くはずですが、例えば「transformKqlでDebug以上をBasicに送ってたまに使う分析用に, Warn以上をAnalyticsに分けてログアラート用にしてコスト最適化しよう！」とすると複数DataFlowになりえるので注意です。

Private Endpointを使う場合は、その分の追加課金が発生する点も忘れずに。

Azure Portal上で試す

ここからは実際にDCRを作成し、ログを送ってみましょう。

事前準備

転送先のLog Analytics Workspaceをあらかじめ作成しておく必要があります。

アクセス制御(IAM)を使いロール割り当てを行う必要があるため、Data Collection Rule (DCR) リソースに対してユーザアクセス管理者あるいは所有者の権限を持っている必要があります。

今回はサービスプリンシパルを使います。

Log Analytics WorkspaceのResource IDを確認

Log Analytics WorkspaceのResource IDを確認しておきます。

Resource IDは命名規則から推測できますし、 Portal や Azure CLI でも確認できます。

以降で使う変数を、まとめて定義しておきます。

# Log Analytics Workspaceのリソースグループ名
RESOURCE_GROUP_NAME="my-resource-group"
# Log Analytics Workspace名
WORKSPACE_NAME="my-workspace"

# 以下は作成に利用

# DCRで送信するさきのテーブル名。"_CL"で終わる必要があります。
WORKSPACE_TABLE_NAME="MyTable_CL"
# DCR名
DCR_NAME="dcr-test"
# DCRのストリーム名。"Custom-"で始める必要があります。
STREAM_NAME="Custom-AllTypes"

# Service Principal名
SP_NAME=sp-dcr-test

Azure CLIでWorkspaceのResource IDを取得します。

# Workspace Resource IDを確認
read WORKSPACE_RESOURCE_ID < <(az monitor log-analytics workspace show --resource-group "$RESOURCE_GROUP_NAME" --workspace-name "$WORKSPACE_NAME" --query "id" --output tsv)

echo $WORKSPACE_RESOURCE_ID

Log Analytics WorkspaceのTableを作成

TimeGenerated (datetime)を含め、出力するカラムをすべて網羅する必要があります。

以下は例です。


az monitor log-analytics workspace table create \
  --resource-group "$RESOURCE_GROUP_NAME" \
  --workspace-name "$WORKSPACE_NAME" \
  --name "$WORKSPACE_TABLE_NAME" \
  --columns TimeGenerated=datetime message=string severity=int sequence=long cpu_ratio=real is_healthy=boolean properties=dynamic request_id=string

なお、カラムの追加はPortal上からあとでも対応できます。

DCR の定義ファイルを作成

定義をJSONファイルで用意します。

今回のスキーマ(columns)は例なので適当ですが、ここに定義したデータしか受け取れないため注意です。もし定義にないカラムを受けた場合は切り捨てられます。

以下を実行してdcr.jsonを作成します。

cat << __EOF__ > dcr.json
{
	"kind": "Direct",
	"properties": {
		"streamDeclarations": {
			"$STREAM_NAME": {
				"columns": [
					{ "name": "message", "type": "string" },
					{ "name": "severity", "type": "int" },
					{ "name": "sequence", "type": "long" },
					{ "name": "cpu_ratio", "type": "real" },
					{ "name": "is_healthy", "type": "boolean" },
					{ "name": "properties", "type": "dynamic" },
					{ "name": "observed_at", "type": "datetime" },
					{ "name": "request_id", "type": "string" }
				]
			}
		},
		"destinations": {
			"logAnalytics": [
				{
					"name": "workspace",
					"workspaceResourceId": "$WORKSPACE_RESOURCE_ID"
				}
			]
		},
		"dataFlows": [
			{
				"streams": ["$STREAM_NAME"],
				"destinations": ["workspace"],
				"transformKql": "source | extend TimeGenerated = ['observed_at']",
				"outputStream": "Custom-$WORKSPACE_TABLE_NAME"
			}
		]
	}
}
__EOF__

各項目の意味は以下の通りです。

streamDeclarations.*: *はdataFlowsのstreamsで指定する名前に対応します
- columns: ストリームの入力スキーマを定義します
destinations.loganalytics.workspaceResourceId: 出力先のWorkspaceを定義します。適宜書き換えます。（テーブル名はdataFlowsで指定します）
dataFlows: streamDeclarationsをどう変換し、どのdestinationsへ出力するか繋げる設定です
- streams: 入力のストリームを選びます
- transformKql: 入力のストリームを変換します。sourceテーブルは入力データであり、streamDeclarationsの入力スキーマに基づくColumnだけを参照できます。TimeGenerated (datetime)が含まれるようにする必要があり、1行で書く必要があります
- destinations: 主にdestinations.logAnalyticsのworkspace名を指定します
- outputStream: Log Analytics Workspaceへ出力するテーブル名を Custom- をプレフィクスに含めて指定します

DCR を作成

Private Endpoint接続を行わない場合、--kind Direct にすることで、Data Collection Endpoint（DCE）を使わずにエンドポイントを生成することができます。

DCRの定義を更新したい場合、dcr.jsonを更新して同じcreateを行うことで上書き更新できますが、即時反映はされません（体感10分ほどかかります）

# monitor-control-service Extensionが必要なため初回はインストールの確認が発生します
az monitor data-collection rule create \
	--resource-group "$RESOURCE_GROUP_NAME" \
	--name "$DCR_NAME" \
	--kind "Direct" \
	--rule-file dcr.json

DCRを作成したらDCRのリソースID、エンドポイント、DCRのImmutable Idを確認してください。

read DCR_ID DCR_ENDPOINT_URL DCR_IMMUTABLE_ID < <(az monitor data-collection rule show  \
	--resource-group "$RESOURCE_GROUP_NAME" \
	--name "$DCR_NAME" \
	--query "{id:id,endponit:endpoints.logsIngestion,immutableId:immutableId}" -o tsv)

echo "DCR_ID: $DCR_ID"
echo "DCR_ENDPOINT_URL: $DCR_ENDPOINT_URL"
echo "DCR_IMMUTABLE_ID: $DCR_IMMUTABLE_ID"

DCRに対して、アクセス制御(IAM) を設定

ログを送信するマシンのマネージドIDやサービスプリンシパルに対して、作成したDCRへMonitoring Metrics Publisherの権限を与えます。

今回はサービスプリンシパルを使います。


# Service Principal を作成
read AZURE_CLIENT_ID AZURE_CLIENT_SECRET AZURE_TENANT_ID < <(az ad sp create-for-rbac --name "$SP_NAME" --skip-assignment --query "{appId:appId,password:password,tenant:tenant}" -o tsv)

echo "AZURE_CLIENT_ID: $AZURE_CLIENT_ID"
echo "AZURE_TENANT_ID: $AZURE_TENANT_ID"

# サービスプリンシパルのオブジェクトIDを得る
SP_OBJECT_ID=$(az ad sp show --id "$AZURE_CLIENT_ID" --query id -o tsv)  
echo "SP_OBJECT_ID: $SP_OBJECT_ID"

取得したプリンシパルのObject IDへDCRに対する"Monitoring Metrics Publisher"ロールを割り当てます。

az role assignment create \
  --assignee-object-id "$SP_OBJECT_ID" \
  --assignee-principal-type ServicePrincipal \
  --role "Monitoring Metrics Publisher" \
  --scope "$DCR_ID"

ログを送るスクリプトを用意する

Log Ingestion APIを利用する必要があるので、Azureのお作法に従いサービスプリンシパルに対応するアクセストークンを取得し、Logs Ingestion APIを利用するシェルスクリプトを実装します。

Gist に書いたのでダウンロードして使います。

curl -LO "https://gist.githubusercontent.com/fukasawah/cabcb38ab0562cbf453b1dd0bb6f5149/raw/bdb3667a8789fc955e9763213c554e6aedf7607b/send-logs-ingestion.sh"

スクリプトでログを送る

現在時刻から1時間前と2時間前のログを模擬して送ります。（日時だけdateコマンドで生成。それ以外はダミー値）

export AZURE_CLIENT_ID=$AZURE_CLIENT_ID AZURE_TENANT_ID=$AZURE_TENANT_ID AZURE_CLIENT_SECRET=$AZURE_CLIENT_SECRET

bash send-logs-ingestion.sh --endpoint-url "$DCR_ENDPOINT_URL" --dcr-immutable-id "$DCR_IMMUTABLE_ID" --stream-name "$STREAM_NAME"  << __EOF__
{"message":"startup complete","severity":4,"sequence":9007199254740991,"cpu_ratio":0.125,"is_healthy":true,"properties":{"app":{"name":"billing-worker","version":"1.2.3"},"deployment":{"region":"japaneast","ring":"canary"},"metrics":{"latency_ms":[12.4,15.8,18.9],"percentiles":{"p50":12.4,"p95":18.9}},"flags":{"dry_run":false,"feature_gates":["nested-dynamic","full-schema"]}},"observed_at":"$(date --iso-8601=ns -d "1 hours ago"| tr , .)","request_id":"3f2504e0-4f89-41d3-9a0c-0305e82c3301"}
{"message":"retry scheduled","severity":2,"sequence":9007199254740992,"cpu_ratio":0.875,"is_healthy":false,"properties":{"app":{"name":"billing-worker","version":"1.2.3"},"retry":{"count":3,"next_delay_seconds":30},"errors":[{"code":"Timeout","transient":true},{"code":"UpstreamBusy","transient":true}],"related":{"tenant":"contoso-prod","resources":[{"type":"vm","id":"vm-01"},{"type":"queue","id":"jobs-primary"}]}},"observed_at":"$(date --iso-8601=ns -d "2 hours ago" | tr , .)","request_id":"8f14e45f-ea6d-46a7-8f7c-5f2d7d9c1b77"}
__EOF__

エラー応答がなく、Sent 2 record(s). と出力されれば送信まではうまくいっています。

エラーの場合はエンドポイントやImmutable IDやStream名が間違っている可能性があります。

Log Analytics WorkspaceのTableを確認する

正常に送られているとログがKQLで確認できます。

ログが確認できない場合、DCRのdataflowのoutputstreamやdestinationのWorkspace IDが間違っている、transformKqlでデータを落としている（whereで絞られている）などの可能性があります。

ログに欠損が見られる場合、入力データのJSONとStreamのスキーマ定義が一致しない、transformKqlでデータを切り捨てている（うっかりprojectで特定のカラムのみ出力したせいで空の値になるなど）、などの可能性があります。

なお、observed_atに過去の日時を指定していますが、受信時刻の2日前～1日後の範囲を超える日時を指定すると、TimeGeneratedは現在日時に置き換わる仕様があります。これはLog Analytics側の仕様なのでDCR固有の話ではなく、Data Collector APIでも同様のはずですが、動作確認時にログの日時が意図せず現在日時になって戸惑ったため、ここに記しておきます。過去のデータをまとめて再投入する場合は範囲に注意してください。

宣伝: Fluentd Output Plugin書きました

fluent-bitはあるらしいのですが、fluentdにはまだなさそうだったので書きました。

初めて作ったFluentdプラグインであるため作法がよくわかっていない、ほとんどを生成AIに作らせていますが、よろしければどうぞ。Azure VMのシステム割り当てマネージドIDで送れることは確認しています。

また、fluent-plugin-azure-loganalyticsには利用面でも実装面でもお世話になっています。ありがとうございます。