Azure Monitor AgentでLinuxのカスタムログをAzure LogAnalyticsに取り込む

# はじめに

* とあるシステムからのログを、Linux Syslogサーバで受け取っていたが、それをAzure LogAnalyticsに転送するまでの手順を共有したい
* 従来は、Log Analytics レガシーエージェントがあったが、2024/8にサポート終了になるということで、Azure Monitor Agent (AMA) での転送が必要になる
* ~~とても複雑な手順になるので、ところどころ辛口コメントを入れさせていただきたい~~
* とてつもなく苦労したため、この記事が誰かの役にたつことを願い。

## 前提となるシステム構成
* VNET内に構築してあるSyslogサーバから、LogAnalyticsにログを転送したい
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/54de66f7-97cb-d348-c902-b2ab0bd7f099.png)

## 取り込みたいカスタムログの概要
* CSV形式でAppサーバーからログが吐き出され、Syslogプロトコル経由でSyslogサーバーに転送され、Syslogサーバー内の、/var/log/remoteフォルダ内に蓄積がなされている。
* 以下の形式。Syslogフォーマットで、MSGフィールドはCSV形式となっている
```
root@AzMonitorAgent01:/var/log/remote# tail test02.log -n 1
Mar 13 15:49:36 AppServer "KDDI Taro","2024-03-13 15:49:10","Allowed","outlook.office365.com:443","CONNECT","200","hoge","fuga"
```

# おおまかな流れ
1. LogAnalyticsで仮のカスタムテーブルを定義する
1. DCE（Data Collection Endpoint）を定義する
1. DCEとSyslogサーバを紐づける
1. Azure Firewallで適切にPolicyを入れる
1. DCR（Data Collection Rule）を定義する
1. いったんログを取り込んでみる
1. データを分解するKQL（Kusto Query Language）を作る
1. KQLの出力と整合のある本番カスタムテーブルを定義する
1. DCRを修正する
1. 完成

これだけ見ても複雑さが想像できると思う。

# 1. LogAnalyticsでカスタムテーブルを定義する
* Azure Portalに入り、以下のPSスクリプトを実行する。貼り付けて実行でOK
* テーブル名は、hogehoge_CLとしてあるので、適宜修正すること。
* AzureサブスクリプションID、リソースグループ名、LogAnalyticsワークスペース名も適宜修正すること。
```powershell
$tableParams = @'
{
    "properties": {
        "schema": {
               "name": "hogehoge_CL",
               "columns": [
        {
                                "name": "TimeGenerated",
                                "type": "DateTime"
                        }, 
                       {
                                "name": "RawData",
                                "type": "String"
                       }
              ]
        }
    }
}
'@

Invoke-AzRestMethod -Path "/subscriptions/xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx/resourcegroups/<リソースグループ名>/providers/microsoft.operationalinsights/workspaces/<LogAnalyticsワークスペース名>/tables/hogehoge_CL?api-version=2021-12-01-preview" -Method PUT -payload $tableParams
```

* 投入後に、StatusCode 200, 設定したテーブル情報が返答されること
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/01a15c8d-ec9d-cbde-8300-a90e2b84abb1.png)



# 2. DCEを定義する
* カスタムログの場合、DCE経由でのログ取り込みが必須となるため、定義する。
* ここは命名だけなので、適宜設定する。
* なお、対象のLogAnalyticsと同じリージョン必須のため注意。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/94a88fc9-5548-d3f7-5600-8783ba869c42.png)

# 3. DCEとSyslogサーバーを紐づける
* 紐づけるだけ。以下よりSyslogサーバーを選択する。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/f5de53e4-7633-0c22-62ce-edf58c60049c.png)
* なお、Syslogサーバー側では、事前にManaged IdentityをONにしておく必要があるので、以下の通りONにしておくこと。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/814c6000-0fc6-71a8-9bd0-2e9086c67b6b.png)

# 4. AzureFirewallにLogAnalytics向けPolicyを入れる
* Syslogサーバーは、LogAnalyticsと通信できる必要があるため、Firewallなど設置している場合は、適宜Allowのポリシーを入れてやる必要がある。
* 今回は、LogAnalyticsをJapanEastリージョンに設置したため、JapanEast向けを設定する
* 以下の宛先、TCP:443宛ての通信をAllowにする。
````
global.handler.control.monitor.azure.com
japaneast.handler.control.monitor.azure.com　※VMの設置されているリージョン。
xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx.ods.opinsights.azure.com　※宛先のLogAnalytics ワークスペースID
hoge-dce-xxxx.japaneast-1.ingest.monitor.azure.com　※設定したDCE
````

# 5. DCRを定義する
* 名前をつけて作る。3.で定義したDCEを指定すること
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/127637d5-f2d8-a39e-846d-70cbc3cedaed.png)

## 5-1. Resourceを指定する。Syslogサーバーを指定して、DCEをEnableする
* なぜはじめの画面にもDCE選択があって、ここにもあるのかは謎。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/def489a9-4619-3deb-37dd-1f802918272c.png)

## 5-2. Collect and Deliverを指定する
* データソースとして、Syslogサーバー内のログが保存されるフォルダ・ファイル名規則を入力する
* テーブル名は最初に定義したテーブル名。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/bc33d437-7be5-0be6-8f07-7fc76c88c79f.png)
* Destinationは取り込み先のLogAnalyticsワークスペースを選択するだけなので適宜。

# 途中確認：AMAがSyslogサーバーにインストールされたか確認する
* ここまで正常に終えると、SyslogサーバーにAMAがインストールされ、Heartbeat通信がLogAnalyticsに転送される
* その確認は、LogAnalyticsのAgentの項目から確認できる。
* 5項まで設定完了して20分経過しても、Heartbeatが流れてこない場合は、どこかで間違っているため、見直すこと。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/3ba6c3cd-5de9-b57c-971d-31fcd410fc5d.png)

# 6. いったんログを取り込む
* DCRで指定したフォルダにログを突っ込むと、LogAnalyticsに取り込みがなされる
* DCRのサーバーへの反映は20分ほどかかるので待つこと。

# 7. データを分解するKQLを作る
* カスタムテーブルのテーブル定義は、1項の通り、RawDataしか定義していないため、取り込んだログは、そっくりRawDataフィールドに格納される
* 以下の内容が、そっくりRawDataに入ることになる。
```
Mar 13 15:49:36 AppServer "KDDI Taro","2024-03-13 15:49:10","Allowed","outlook.office365.com:443","CONNECT","200","hoge","fuga"
```
* このRawDataから、CSV部分を切り出し、CSVをParseして、カスタムログの各フィールドに分解してやりたい。
* 以下の例となる。
    * 2行目で、Syslogのヘッダ部分を削除している
    * 3行目・4行目で、DCR内のKQLでは、なぜかparse_CSV関数が使えないため、仕方なくtrimとsplitを組み合わせて、CSVをParseしている。
    * 6行目は、JSTでログが吐き出されているが、DCRで取り込まれる際、時刻がUTCと認識されるため、JSTをUTCに変換するため、マイナス9時間している
```
hogehoge_CL
| extend TrimData = trim_start(@"^((?:.*? ){4}.*?)", RawData)
| extend TrimData2 = trim('"',TrimData)
| extend CSVFields_CF = split(TrimData2, '","')
| extend User = tostring(CSVFields_CF[0])
| extend EventTime = datetime_add('hour', -9, todatetime(CSVFields_CF[1]))
| extend Action = tostring(CSVFields_CF[2])
| extend Host = tostring(CSVFields_CF[3])
| extend Method = tostring(CSVFields_CF[4])
| extend RespCode = tostring(CSVFields_CF[5])
| extend FileType = tostring(CSVFields_CF[6])
| extend HogeType = tostring(CSVFields_CF[7])
| project TimeGenerated, EventTime, User, Action, Host, Method, RespCode, FileType, HogeType
 ```

## KQL作りのTips
* letで生ログを1行与えて試すことができる。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/710b7944-67b4-f36b-fc18-232a1514374a.png)


 # 8. KQLの出力結果に合わせて本番カスタムテーブルを定義する
 * 7項で以下のフィールド定義としている。
 ```
 project TimeGenerated, EventTime, User, Action, Host, Method, RespCode, FileType, HogeType
 ```
* Azure Portalから、Edit Schemaより修正、または1項のようにPowerShellでもよい。
 ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/22daa97e-d102-dc10-7b58-cb82102cc9ae.png)

* CSV Parseの際に設定した型と合わせること。例はDateTime型とString型。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/a9312ea3-b908-4ee9-f76b-ca7d2278a796.png)

# 9. DCRを修正する
* ログ取り込みと同時に分解するためのTransformationに先ほど作ったKQLを入れる
* この際、テーブル名は source となるため注意。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/10317/1f17ad8b-19ca-79f5-9e30-6338f58f8921.png)

# 完成
* 反映までに20分ほどかかる
* ログがしかるべきテーブルカラムに分解されてLogAnalyticsに取り込まれることを確認しよう
* 仮のテーブルは適宜削除しよう


# 参考にした記事
![Windows テキストログを Azure Monitor Agent を用いて、カスタムログとして Microsoft Sentinel (LogAnalytics) に基本ログで取り込む！](https://qiita.com/hisnakad/items/c6e83e3989509e9401fb)