初めに

この記事はバリバリのSIEM/SOAR関連のエンジニア向けではなく、業務システム寄りのIT担当者向けです。

間違いのご指摘は大歓迎です。なので継続的に更新するかもしれません。

(この記事は2024年4月末時点の情報をもとにしています)

CSVファイルをLog Analyticsのカスタムテーブルに読み込ませる方法

任意のCSV形式のログファイルを、Log Analyticsのカスタムテーブルに読み込ませる(＝インジェストする)方法について説明します。

Log Analyticsの概要についてはすでにご存じであるという前提です。

カスタムテーブルとはLog Analyticsのワークスペース内に自由なカラム定義で作成できるテーブルです。

インジェストしたいCSV形式のログファイルの内容にあわせて、カスタムテーブルを作成することになります。

Log Analyticsに読み込ませる理由

数億行単位のCSV形式のログファイルがあるとします。

仮に、リレーショナルデータベースに一括インポートして実用的な検索速度を出すには、かなりのチューニングが必要です。

一方、Log Analyticsにインジェストすればチューニングなしで実用に耐える検索速度が出ます。しかもSQL文とおおむね同レベルの柔軟な検索ができます。

もちろん同じことはLog Analyticsと同じカテゴリの別製品でもできますが、筆者に知識がないのでここではMicrosoftのAzure環境を前提とします。

カスタムテーブルを作成する準備

Log Analyticsにワークスペースを作る

説明略。すでに完了している前提とします。

CSVファイルを置くためのフォルダを用意する

インジェストしたいCSVファイルを置くためのフォルダを用意しましょう。

Azure仮想マシン上のフォルダを使う前提とします。Log Analyticsを使える環境があるのにAzure仮想マシンを作れないことはないはずだからです。

最終的に、そのフォルダに一定の命名規則で作成したCSVファイルをつぎつぎ書き出すと、「Azure Monitorエージェント」が自動でLog Analyticsのカスタムテーブルにインジェストしてくれる、という仕組みが出来上がります。

「Azure Monitorエージェント」がそのフォルダを常に監視しており、新しいファイルが現れたらすぐ読み込んでくれるイメージです。インジェストし終えたCSVファイルが重複して読み込まれることはありません。

JSON形式のサンプルデータを用意する

CSVファイルをインジェストしたいのに、なぜわざわざJSON形式のサンプルデータが必要なの、という疑問が湧きますがそう決まっているので仕方ありません。

JSON形式のサンプルデータが必要な理由

サンプルデータはLog Analyticsのカスタムテーブルの定義を作成するためのものですが、カスタムテーブルに「データ収集ルール(DCR)」という方式でCSVファイルをインジェストするには、サンプルデータをJSON形式で用意する必要があります。

「データ収集ルール(DCR)」ではない古いインジェスト方式(MMA)ならサンプルデータはCSV形式でよかったのですが、このMMAは2024年8月に終了となっています。

"TimeGenerated"という列が必要

また、JSON形式のサンプルデータには"TimeGenerated"という名称のdatetime型の列が必要です。別の名前ではいけません。この列がないとサンプルデータをアップロードしたときエラーになります。

それに関連する話ですが、インジェストするCSVファイルにヘッダ行を入れてはいけません。Log Analyticsでインジェストする際、ヘッダ行もログデータとして読み込まれてしまいます。1行目からいきなりログデータ部になっている必要があります。

ではインジェストするとき、CSVファイルの各列がカスタムテーブルのどのカラムに対応するのかをどう判断するのか、となります。

たとえば、Log Analytics側でCSVファイルの列に左から0、1、2という番号をつけて、カスタムテーブルの各列も列番号を持っているのか等々となりますが、違います。

これは最後のほうで説明します。

JSON形式のサンプルデータをPowerShellで作成

ではCSV形式のログファイルからJSON形式のサンプルデータを作成します。

PowerShellやPythonで作成すればいいだけなので説明は不要だと思いますが、例としてPowerShellのスクリプトを書いておきます。

CSVファイル内にあるタイムスタンプ列を残しつつTimeGenerated列を追加するパターンです。CSVファイルの最初の20行をサンプルデータにするとします。

Import-Csv -Path 'C:\temp\InputFile.csv'
| Select-Object -First 20
| Select-Object -Property *,
    @{ Name       = 'TimeGenerated';
       Expression = {(Get-Date '2024-01-01 00:00:00')} }
| ConvertTo-Json
| Out-File -FilePath 'C:\temp\OutputFile.json'

サンプルデータなのでTimeGenerated列に入れる値は何でもいいです。ここでは固定で'2024-01-01 00:00:00'にしてみました。

CSVファイル内にあるタイムスタンプ列をそのままTimeGenerated列に変換するためのスクリプトはご自身で考えてください。(面倒なので丸投げ)

作成したサンプルデータをアップロードするのはまだまだ先です。

Azure仮想マシンにAzureモニターエージェントをインストールする

ログの置き場にするAzure仮想マシンに「Azureモニターエージェント」をインストールします。

と、書いておきながら、Azureモニターエージェントをインストールする必要はありません。後述の「データ収集ルール」を作成すると勝手にインストールされます。

便利ですね。

データ収集エンドポイントを作成する

カスタムテーブルを作成する準備はまだ続きます。

「データ収集エンドポイント」というものを作成します。ざっくり言えば、CSVファイルの置き場にするAzure仮想マシンを指定するためです。

データ収集エンドポイントの作成

Azureポータル ( https://portal.azure.com/ ) から「モニター」を開く
左端の下の方にある「データ収集エンドポイント」をクリック
左上の「＋作成」をクリック
「エンドポイント名」は適当に入力 (当然、命名規則を決めておいた方がいいです)
「サブスクリプション」には、自組織のサブスクリプション名が入っているはず
「リソースグループ」には、CSVファイルをインジェストするLog Analyticsワークスペースと同じリソースグループを指定

次の「タグ」画面でのタグ付けは必須ではありません。管理目的でタグ付けしたければしてください。

以上で「確認と作成」をクリックします。

データ収集エンドポイントの中身の設定

ここからデータ収集エンドポイントの中身を設定していきます。

「モニター | データ収集エンドポイント」画面のデータ収集エンドポイント一覧から、上記で作成したデータ取集エンドポイントをクリック
データ収集エンドポイントの名前で画面が開くので、左端の「リソース」メニューをクリック
左上の「＋追加」をクリック
右側に「範囲の選択」という画面が出てくるので、CSVファイル置き場にするAzure仮想マシン名を検索して「適用」をクリック

これで「データ収集エンドポイント」の作成と設定が終わりました。

カスタムテーブルを作成する

いよいよLog Analyticsワークスペースにカスタムテーブルを作成します。

下記お読みいただくと分かりますが、カスタムテーブル作成の途中で「データ収集ルール」の作成に寄り道します。

なぜこんなややこしい流れになるのかは不明です。

カスタムテーブルの作成

カスタムテーブルを作成するLog Analyticsワークスペースを開く
左端にある「テーブル」メニューをクリック
左上の「＋作成」をクリックして「新しいカスタムログ (DCRベース)」を選択

「DCR」は「Data Collection Rule(データ取集ルール)」の意味です。

次に「カスタムログを作成」という画面に推移します。

「カスタムログを作成」

この「カスタムログを作成」画面では、カスタムテーブルの基本情報と、読み込ませるCSV形式ファイルに関する情報を設定します。

「カスタムログを作成」の「基本」タブ

「テーブル名」をつける
カスタムテーブル名は必ず「_CL」で終わりますが「_CL」部分の入力は不要です
「データ収集ルール」で「新しいデータ収集ルールの作成」をクリック

ここからいったん「データ収集ルール」の作成に寄り道します。

ややこしいですね。

カスタムテーブルを作成するにはデータ収集ルールが必要で、データ収集ルールの作成にはカスタムテーブルが必要という、一見すると「にわとりとタマゴ」状態になっています。

実際にはここでデータ収集ルールの概要だけを作成し、後で詳細な中身を設定する手順になります。

「新しいデータ収集ルールの作成」の概要を作成

上記で「新しいデータ収集ルールの作成」をクリックすると、画面の右側に「新しいデータ収集ルールの作成」という画面が出てきます。

「サブスクリプション」には自組織のサブスクリプション名が入っているはず
「リソースグループ」には、CSVファイルをインジェストするLog Analyticsワークスペースと同じリソースグループを指定
「地域」は「リソースグループ」にもとづき自動で値が入る
「名前」に適当な名前を入力 (当然、命名規則を決めておいた方がいいです)
「完了」をクリック

これで「データ収集ルール」の概要の作成は終わりで、後で詳細な中身を設定します。

「カスタムログを作成」の「基本」タブ(続き)

データ収集ルールの概要の作成が終わると「カスタムログを作成」画面の「基本」タブに戻ってくるので、続きを入力します。

「データ収集エンドポイント」は上記で作成した「データ収集エンドポイント」を選択
「次へ」をクリック

「カスタムログを作成」の「スキーマと変換」タブ

ここでようやく、最初に作成したJSON形式のサンプルデータの出番になります。

「サンプルデータのアップロード」をクリックし、JSON形式のサンプルデータをアップロードすると表形式でログが表示されます。

想定どおりの列名とデータになっていることを確認して「次へ」をクリックします。

なお「スキーマと変換」タブの左上に「変換エディター」というものがあります。

これをクリックすると、元のCSVファイルの列定義とカスタムテーブルの列定義が異なる場合、どのように対応づけるのかのロジックをKustoクエリー言語(KQL)で書けるようになっています。

インジェストするときにリアルタイムで列の並びやデータの中身を変換するイメージです。

この記事ではこの「変換エディター」部分はスルーし、あとで「Transform」という入力欄にKQLで変換規則を設定することにします。

KQLで書き切れないほど変換規則が複雑な場合は、事前にCSVファイルを別のCSVファイルにスクリプトで変換しておく方法もありそうです。

カスタムテーブルのスキーマ定義の変更方法

作成したカスタムテーブルのスキーマ定義はいつでも変更できます。その方法は以下のとおり。

カスタムテーブルを作成したLog Analyticsワークスペースを開く
左端のメニューで「テーブル」をクリック
テーブル一覧からカスタムテーブルを見つけて右端の「・・・」をクリック
「スキーマの編集」をクリック
右側に「スキーマエディター」が開くので必要な列を追加し「保存」をクリック

列の順番はCSVファイルの列と同じ並びでなくてもいいです。その理由は後述します。

上半分には内部的に課金管理などで利用される列がいくつか表示されていますが無視しましょう。

また、じつは"RawData"という非常に重要な列があるのですが、いわば「隠しカラム」なので表示されません。これについては後述します。

「データ収集ルール」の中身を設定する

先ほどカスタムテーブルを作成する途中で「データ収集ルール」の概要だけを作成しましたので、その中身を設定していきます。

Azureポータル ( https://portal.azure.com/ ) から「モニター」を開く
左端の下方にある「データ収集ルール」をクリック
データ収集ルールの一覧に先ほど作ったルール名があるので、それをクリック

するとそのデータ収集ルールの名前で画面が開きます。

「リソース」の設定

まずデータの収集元になるリソースを設定します。CSVファイルを置くAzure仮想マシンのことです。

さっき「データ収集エンドポイント」でも設定したのに、またここで設定が必要な理由はよくわかりません。

左側の「リソース」をクリック
左上の「＋追加」をクリック
右側に「範囲の選択」画面が出てくるので、CSVファイルを置くAzure仮想マシンを選択して「適用」をクリック

「データソース」の設定

この「データソース」の設定がデータ収集ルールの設定で最も重要な部分です。

左側の「データソース」をクリック
データソースの一覧に「カスタムテキストログ」という項目があるはずなので、それをクリック

すると画面右側に「データソースの追加」画面が開きます。

「データソースの追加」の「データソース」タブ

「データソースの種類」は「カスタムテキストログ」になっているはず
「ファイルパターン」は以下に解説

この「ファイルパターン」にはAzure仮想マシン上のCSVファイルのフルパスをワイルドカードで指定します。

たとえば「D:\CSVLog\user_operation_detail*.csv」などです。正規表現は使えないようです。

「D:\CSVLog*.csv」のようなざっくりしたパターンだと、間違って無関係なCSVファイルを書き込んでもインジェストされてしまいます。できるだけ限定的な命名規則にするのがいいでしょう。

「テーブル名」に上記で作成したカスタムテーブル名を入力します
ここでは末尾の「_CL」まで入力する必要があります
「レコードの区切り記号」は「End-of-Line」になっているはず
「Transform」には「source」とだけ書いてあるはず

この「Transform」が先ほど「変換エディター」のところで言及したKQLをつかった変換規則です。

「Transform」の書き方

ここから「Transform」についてやや長い説明になります。

この「Transform」(データ変換規則)は個人的に直観に反するものでした。

CSVファイルの列の並びに合うように、カスタムテーブルの方もカラムを並べておけば、変換規則なしでもそのままインジェストしてくれそうな気がします。

業務システムを扱っていると、CSVファイルと読み込み先テーブルの列の並びは一致させるのが普通でしょう。

しかしLog AnalyticsではCSVファイルをインジェストするとき、CSVファイル1行分のデータが丸ごと「RawData」という列に読み込まれます。先述のように「スキーマエディター」に現れない、いわば隠しカラムです。

その後に「RawData」列に入ったのデータを、KQLのparse演算子で解析してカスタムテーブルの各カラムにバラす必要があります。

どうしてこんな面倒なことをするのかと最初は思いましたが、この手順を踏めば、CSVファイルとカスタムテーブルの列の並びが違っていても問題ありません。

というより、Log Analyticsのカスタムテーブルのカラムには「順番」がありません。

先述のように、CSVファイルをLog Analyticsのカスタムテーブルにインジェストするとき、CSVファイルのヘッダ行は無意味であり、ヘッダ行を書いてはいけません。

その代わりこの「Transform」(変換規則)で、CSVファイルの列を左から順に機械的にカスタムテーブルのどのカラムに対応させるかを記述します。

たとえば元のCSVファイルに左から、ログ日時、ユーザー名、フルパスに当たるデータが入っている場合、以下のようなKQLで書いた変換規則を「Transform」欄に入力します。

(KQLのparse演算子の詳細は公式ドキュメントなどをお読みください)

source
| parse RawData with '"' TimeGenerated:datetime '","' UserName:string, '","' FullPath:string '"'

この変換規則はCSVファイル内の列を、左から機械的にカスタムテーブルのTimeGenerated、UserName、FullPathにインジェストするという変換規則です。Log AnalyticsがCSVファイルの各列の「意味」を解釈しているわけではありません。

これと同じ情報を持つCSV形式のログファイルが左からユーザー名、ログ日時、フルパスにあたるデータの並びになっている場合、変換規則はこうなります。

source
| parse RawData with '"' UserName:string '","' TimeGenerated:datetime, '","' FullPath:string '"'

なお、どれだけ長いKQLを作成しても「Transform」入力欄の1行に詰め込まれるので、画面上でKQLを確認することは事実上不可能ですw

KQLで表現できさえすれば、いくらでも複雑な変換規則を書くことができます。

「データソースの追加」の「ターゲット」タブ

「Transform」にKQLを入力し終えたら次の「ターゲット」タブに移ります。

「ターゲットの種類」は「Azure Monitor Logs」を選択
「サブスクリプション」は自組織のサブスクリプション名になっているはず
「アカウントまたは名前空間」はカスタムテーブルを作成したLog Analyticsワークスペースを選択
最後に「保存」をクリック

実際のCSVファイルのインジェスト

CSVファイルを置くために用意したフォルダへ、「ファイルパターン」で指定したパターンに沿ったファイル名で、「Transform」の変化規則でカスタムテーブルの正しいカラムに読み込まれる列の並びになっているCSVファイルを書き出していけば、順番にインジェストしてくれます。

最後に

いろいろ書き漏らしているような気がしますが、とりあえずは第一稿ということで。

くり返しになりますが、間違いがあれば遠慮なくご指摘ください。