Azure Log Analytics のデータを Blob Storage と Azure Data Explorer に退避してみた - 2年以上ログを検索できるようにしたいときに

本記事について

本記事では、ログの保持期間が最長2年間である Log Analytics の中のデータをより長い期間保存し、さらに同じ検索言語である Kusto Query Language (KQL) でサーチしたいときに利用できる Azure Data Explorer をご紹介していきます。たとえば Azure Sentinel で取得した Office 365 の監査ログを 5年や10年保存しておきたい時などに便利です。

具体的に、Log Analytics から Data Explorer にデータを退避し、かつ Log Analytics と同じクエリを利用できることを見ていきます。

また、Log Analyticsのデータの長期保管については、こちらの記事にてオプションをまとめてみましたので合わせてご覧ください。

Azure Data Explorer について

Azure Data Explorer の概要については、公式ウェブサイトのdocsのほか、こちらの Qiita 記事でも詳細にご説明されていらっしゃいます。Data Explorer はログと利用統計情報データのための、高速で拡張性に優れたデータ探索サービスです。検索には、Kusto Query Language (KQL)を利用します。

また、KQL は、Azure Log Analytics や Azure Application Insights, Microsoft Defender ATP でも利用されているマイクロソフトのクエリ言語で、Azure を利用する際にはマスターしておくと便利です。

Log Analytics から Data Explorer への退避と検索

1. Log Analytics からのデータの取得

Log Analytics は API を提供しており、API 経由で検索してデータを取得することができます。Qiita では、こちらの記事が詳細な手順をご説明くださっており、本記事ではそのまま利用します。

なお、API には制限があるので、ご注意ください。

Query limits
As well as call rate limits and daily quota caps, there are also limits on queries themselves:
Queries cannot return more than 500,000 rows
Queries cannot return more than 64,000,000 bytes (~61 MiB total data)
Queries cannot run longer than 10 minutes by default. See this for details.

Rest API のリクエストの中身は、下記のようなサーバーの Heartbeat の取得にしてみます。

let startdatetime=datetime(2020-04-02T00:00:00);
let enddatetime=datetime(2020-04-02T23:59:59);
Heartbeat | where TimeGenerated between(startdatetime .. enddatetime)

$body = @{query = 
'let startdatetime=datetime(2020-04-02T00:00:00);
let enddatetime=datetime(2020-04-02T23:59:59);
Heartbeat | where TimeGenerated between(startdatetime .. enddatetime)
'
}| ConvertTo-Json

具体的には、こんな感じの CSV のファイルを取得できます。

2. Azure Data Explorer のクラスターの作成

こちらの Qiita 記事 の手順に従って Data Explorer クラスターとデータベースを作成していきます。

クラスターは、ここでは Dev を選択しました。

データベースも作成します。

3. Azure Data Explorer へのデータのアップロード

次にデータベースにテーブルを用意します。概要ページの URI にブラウザの別タブでアクセスします。

すると下記のようなページが出てきます。こちらが Data Explorer でデータを扱うためのサイトになります。

ここで、さきほど作成したデータベースにテーブルを作っていきます。データベースを右クリックして、Ingest new data (preview) を選択します。

テーブル名を Heartbeat として、Ingestion type を from file としてさきほど取得した CSV ファイルをアップロードします。

次にスキーマの設定に進みます。ここでは TimeGenerated を datetime に変更します。

そして、Start ingestion をクリックすると、データが取り込まれます。試しに、Heartbeat と打って実行すると、Log Analytics と全く同じ要領で検索ができます。

4. Azure Storage からのデータの取り込み

次に、Azure Storage に入った CSV を検索できるようにしていきます。

まず、Azure Storage と Event Hub / Event Grid を用意します。
Storage については、Storage Account と Blob コンテナーをひとつずつ用意します。

Event Hub については、Event Hub 名前空間を作り、そのなかに Event Hub をひとつ作ります。

次に Event Grid サブスクリプションを作っていきます。Event Grid のサブスクリプションでは、さきほど作った Storage Account をトピック、Event Hub をエンドポイントとして作成していきます。

これで準備完了です。Azure Portal の Data Explorer クラスターのページに戻り、データベースに進んで作ったデータベースをクリックします。（ここからの手順の詳細については、公式ドキュメントもご参照ください。）

そして、データ接続の追加をクリックします。

ここでは接続を Blob ストレージとして、イベントグリッドをさきほど作成したサブスクリプション、テーブルを Heartbeat, データ形式を CSV にします。そして作成します。

これでストレージから Data Explorer への取り込みの準備は完了です。あとは、Heartbeatの検索結果で、さきほどとは TimeGenerated の条件を変えた別のCSVファイルを用意して、Blobストレージのコンテナーにそれをアップロードします。

すると、新しいデータも Heartbeat テーブルに格納されます。この仕組みを使って、継続的に Log Analytics から Storage にデータを退避させると、新しいデータを Data Explorer で検索することができます。

5. Azure Data Explorer での検索

Log Analytics での検索と全く同じクエリを Data Explorer でも使うことができます。
試しに以下のクエリを試してみます。

let startdatetime=datetime(2020-04-03T00:00:00);
let enddatetime=datetime(2020-04-03T23:59:59);
Heartbeat 
| where TimeGenerated between(startdatetime .. enddatetime)
| summarize count() by bin(TimeGenerated, 1h), Computer
| render timechart 

• Log Analytics での結果

• Data Explorer での結果

若干 UI は違いますが、同じクエリを使うことができることが確認できます。

最後に

本記事では、Log Analytics から Data Explorer にデータを退避し、同じクエリを利用できることを確認していきました。実際に退避をする際には、Azure Functions などを利用して、継続的にログを送信する仕組みを利用すると、ダウンロード・アップロードも自動化できます。

＊本稿は、個人の見解に基づいた内容であり、所属する会社の公式見解ではありません。また、いかなる保証を与えるものでもありません。正式な情報は、各製品の販売元にご確認ください。