OpenLDAPのログをElastic AgentのCustom Logsで取り込む

# はじめに
Elasticsearchでログを取り込むと言えばBeatsかLogstash、という方も多いと思いますが、Kibanaから集中管理できるElastic Agentが広く使われています。
Elastic Agentについてはこちらの記事をご参照ください。
https://qiita.com/yukshimizu/items/abec12ac46749db46b40

BeatsやLogstashでは取得側でパース等の設定をしていましたが、Elastic AgentではIngest Pipelineを使用してElasticsearch側でパースを行います。
Ingest Pipelineについてはこちらに正式ドキュメントがあります。
https://www.elastic.co/guide/en/elasticsearch/reference/8.11/ingest.html

本記事ではElastic AgentのIngegrationsが用意されていないOpenLDAPのログをGrokを用いてIngest Pipelineでパースを行ってインデックス手順を解説いたします。
因みにIntegrationsというのは、BeatsのModuleのようなもので、ApacheやNGINXなどログフォーマットが予めわかっているログをパースするIngest Pipelineが事前に用意されていて、Dashboard等も自動で作成する機能です。

以下の内容になっていますので、OpenLDAPのログがすでにある場合にはOpenLDAPの設定などは飛ばしてください。
1. OpenLDAPとログイン用アプリ設定
1. Elastic Agent設定・インストール
1. Ingest Pipeline設定

# Elastic Cloudにクラスタを作成
（既にある場合は必要ありません）
こちらのブログをベースにElastic Cloudにクラスタを展開します。
https://qiita.com/tomo_s_el/items/3584d0b1fabb0bafa4fa

# 動作環境
Dockerが動けば特に環境は問いません。
OS: Linux (ubuntu 20.04)
VM: Azure Standard DS1 v2
Docker: 24.0.5
Docker Compose: v2.20.2
Elasticsearch (Elastic Cloud): 8.11.1

# OpenLDAPとログイン用アプリ設定
まずOpenLDAPのログを作るためにOpenLDAPを立てて、認証を行う簡単なアプリを立ち上げます。
docker composeで立ち上げます。
こちらのリポジトリにソースコードがあります。
https://github.com/legacyworld/customlogs_ldap
基本的に何も設定は必要ありません。

## OpenLDAP
OpenLDAPのDocker Imageはこちらを利用しています。
https://github.com/osixia/docker-openldap
以下の`Dockerfile`でbuildします。
```Dockerfile
FROM osixia/openldap
RUN apt-key adv --keyserver keyserver.ubuntu.com --recv-keys 0E98404D386FA1D9
RUN apt-get update && apt-get install -y rsyslog
COPY user.ldif /container/service/slapd/assets/config/bootstrap/ldif/custom
COPY logging.ldif /container/service/slapd/assets/config/bootstrap/ldif/custom
COPY rsyslog.conf /etc
```
- Ubuntuの`apt-key`
    - こちらを参考にしました
    - https://zenn.dev/takc923/articles/391ea8a4e03f43
- ldifファイル
    - ldifファイルを指定されたフォルダにおいておくと起動時に設定してくれます
    - `user.ldif`
        - testuser{1..4}を追加しています
    - `logging.ldif`
        - syslogにログを出力する設定
- rsyslog.conf
    - 時間のフォーマットを変更(usまで表示)

OpenLDAPのadminアカウントは`docker-compose.yml`で設定します。
```docker-compose.yml
    environment:
      - LDAP_LOG_LEVEL=128
      - LDAP_DOMAIN=example.com
      - LDAP_ADMIN_PASSWORD=password
```
## ログインアプリ
こちらを参考にしました。
https://www.osstech.co.jp/~hamano/posts/flask-login-ldap/
元記事のソースコードからは少し変更しています。
- importの変更
    - `From`から`FlaskForm`に変更
    - `TextField`から`StringField`に変更
    - `logging`追加
- LDAPの設定
    - LDAP URLをコンテナ名に変更
    - DNを`user.ldif`の設定に合わせて変更
- logging
    - `/log/flask.log`に出力するように変更
## 起動
`customlogs_ldap`ディレクトリで`docker compose up -d`
立ち上がったらOpenLDAPのコンテナ上でrsyslogを起動します。
`docker exec -it openldap service rsyslog start`

## UIからの認証
http://localhost:5000
で以下の画面が開きます。
![スクリーンショット 2023-11-30 11.55.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/59c7ad98-ba14-424d-f532-ed63d87da65c.png)
testuser1/passwordと入力すると以下のように表示されます。
![スクリーンショット 2023-11-30 15.35.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/2f19d23d-4dfa-2584-b0db-9d4b690bc466.png)
パスワードを間違えるとページは遷移しません。
`log/syslog`には以下のように表示されています。
```
2023-11-30T02:58:06.927861+00:00 1fb5472bef24 slapd[441]: conn=1000 fd=12 ACCEPT from IP=192.168.48.3:54620 (IP=0.0.0.0:389)
2023-11-30T02:58:06.928033+00:00 1fb5472bef24 slapd[441]: conn=1000 op=0 BIND dn="uid=testuser1,ou=people,dc=example,dc=com" method=128
2023-11-30T02:58:06.928162+00:00 1fb5472bef24 slapd[441]: conn=1000 op=0 BIND dn="uid=testuser1,ou=people,dc=example,dc=com" mech=SIMPLE ssf=0
2023-11-30T02:58:06.928283+00:00 1fb5472bef24 slapd[441]: conn=1000 op=0 RESULT tag=97 err=0 text=
2023-11-30T02:58:06.929043+00:00 1fb5472bef24 slapd[441]: conn=1000 op=1 UNBIND
2023-11-30T02:58:06.929547+00:00 1fb5472bef24 slapd[441]: conn=1000 fd=12 closed
```
本記事ではこのログをパースしていきます。

## スクリプトでログ生成
`docker exec -it app python test.py`を実行するとランダムにLogin IDを選び、10%ほどでパスワードを間違えます。

# Elastic Agentの設定
## Integrationの設定
ここではログの絶対パスを設定するだけです。
KibanaメニューのManagement -> Integrationsとクリックします。

検索バーに**custom logs**と入力し、赤枠で囲った部分をクリックします。
![スクリーンショット 2023-11-30 22.49.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/d7b92189-7af6-dc62-0d3e-cdc82ce4dc9c.png)
右上の**Add Custom Logs**をクリック
![スクリーンショット 2023-11-30 22.56.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/853c8259-55fb-00e7-143e-caadf958afd1.png)

初回のみこの画面が開くこともあります。Add Integration only(skip agent installation)をクリックします。
![スクリーンショット 2023-12-01 10.45.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/17dc9cb3-90f2-1bb7-a71f-738bad408b1a.png)

次の画面では2箇所入力が必須です。
- Log file path
    - customlogs_ldap/log/syslogの絶対パスを記入
- Dataset name
    - インデックスの名前に使用します。ここでは**ldap**と入力
最後に右下の**Save and Continue**をクリック
![スクリーンショット 2023-11-30 23.01.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/48337391-36dc-9193-b371-b0cf0fe3e586.png)

暫くすると以下のダイアログが出てきますので、**Add Elastic Agent to your hosts**をクリック
![スクリーンショット 2023-11-30 23.05.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/eb0d2b24-9b59-3336-b635-49ad3d02a371.png)

## Elastic Agentのインストール
以下のような画面が表示されるので、**Install Elastic Agent on your host**の部分でOpenLDAPをインストールした環境を選択する（下記のキャプチャではLinuxへのインストール例）
![スクリーンショット 2023-11-30 23.14.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/13185ffc-a67b-3347-4d25-2818f9e4ec46.png)

問題なければ**Confirm agent enrollment**のところが以下のようになります。
![スクリーンショット 2023-12-01 9.24.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/28305db8-3d2f-b8a1-a6c3-473801a6c255.png)

Management -> Fleetとクリックすると、以下のように登録されていることがわかります（Hostのところの名前は各自環境で変わります）
もう一つ表示されていると思いますが（下記だとd714a2d93a83）、これは気にしなくても良いです。
![スクリーンショット 2023-12-01 11.09.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/80f4d8b2-c23c-be03-2d6d-ed34e28ba5d1.png)

## データの確認
データが取り込まれているので確認してみます。
Management -> Stack Management -> Kibana -> Dataviewをクリックします。
**Dataview**はどのインデックスを処理するか、という単位です。
いろんなログファイルを一括で処理したい場合、**logs**というDataviewで`*logs*`にマッチするインデックスを纏めて処理する、ということが出来ます。
今回は**logs-ldap-default**という名前で自動的にインデックスが作成されています。
Elastic Agentで取得したログの命名規則は下記のページの参照してください。
https://www.elastic.co/guide/en/fleet/8.11/data-streams.html#data-streams-naming-scheme

下記のページで右上の**Create data view**をクリックします。
![スクリーンショット 2023-12-01 12.46.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/bdaee7c8-3f79-1341-fd7d-5a0b24602766.png)

以下のように入力します。
- Name
    - ldap (これは任意です)
- Index pattern
    - logs-ldap*
        - logs-ldapから始まるインデックス全て表示するという設定

右側に**logs-ldap-default**と表示されていることを確認して、**Save data view to Kibana**をクリックして保存します。
![スクリーンショット 2023-12-01 12.54.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/468e3f44-8ec2-9f14-7fdd-9b8440dcc2ad.png)

では実際にデータを見てみましょう。
Analytics -> Discoverをクリックします。先程作成したDataviewの名前(ここでは**ldap**)を選ぶと右側にログが表示されます。
![スクリーンショット 2023-12-01 14.51.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/d93e37e2-b75c-2d47-2da2-6a04dfa20d2b.png)

更に右側の赤枠で囲っている矢印をクリックするとログの詳細を見ることが出来ます。
![スクリーンショット 2023-12-01 14.52.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/1750c99a-b591-2d95-030d-5fbfa7a114c4.png)

これを見ると既にパースされているように見えますね？
でも残念ながら違います。ここにあるIPアドレスやホスト名などはElastic Agentが自動的に収集するデータで、OpenLDAPのログは含まれていません。
OpenLDAPのログは**message**というフィールドに入っています(右下の**2**をクリックすると表示されます)
![スクリーンショット 2023-12-01 14.55.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/c5719328-41d9-ee8a-8bd3-4aa40cf5e7bd.png)
このままだと**uid**や**ou**でグラフ化したり出来ません。
ではこれをパースするingest pipelineを作成していきます。

# Ingest Pipeline作成
**message**フィールドに入っているデータからconnection id,uid,ouといった値を個別に取り出すのはIngest Pipelineの役割です。
https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html
このPipelineの**Grok**を使います。
https://www.elastic.co/guide/en/elasticsearch/reference/current/grok-processor.html
Logstash等で慣れている方も多いと思いますが、空白や文字などで区切っていくものです。
では早速以下のログを全てパースできるようにしていきます。
```
2023-11-30T02:58:06.927861+00:00 1fb5472bef24 slapd[441]: conn=1000 fd=12 ACCEPT from IP=192.168.48.3:54620 (IP=0.0.0.0:389)
2023-11-30T02:58:06.928033+00:00 1fb5472bef24 slapd[441]: conn=1000 op=0 BIND dn="uid=testuser1,ou=people,dc=example,dc=com" method=128
2023-11-30T02:58:06.928162+00:00 1fb5472bef24 slapd[441]: conn=1000 op=0 BIND dn="uid=testuser1,ou=people,dc=example,dc=com" mech=SIMPLE ssf=0
2023-11-30T02:58:06.928283+00:00 1fb5472bef24 slapd[441]: conn=1000 op=0 RESULT tag=97 err=0 text=
2023-11-30T02:58:06.929043+00:00 1fb5472bef24 slapd[441]: conn=1000 op=1 UNBIND
2023-11-30T02:58:06.929547+00:00 1fb5472bef24 slapd[441]: conn=1000 fd=12 closed
```
Grokでは前から順に区切っていくのが楽です。全てに共通している`2023-11-30T02:58:06.927861+00:00 1fb5472bef24 slapd[441]: conn=1000`をまず切ります。
Management -> Stack Management -> Ingest Pipelineを選び、右上の**Create pipeline**から**New pipeline**をクリックします。
![スクリーンショット 2023-12-01 15.31.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/3dbcf8cb-2930-4569-8fdc-140d542856bd.png)
以下の画面が開きます。
![スクリーンショット 2023-12-02 17.24.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/ce3ca977-6913-c3da-b7a1-e6363640e328.png)
- Name
    - ldapと入力（任意です）

## 1つ目のGrok
真ん中にある**Add a processor**をクリックすると右側に以下の画面が出てくるので設定をしていきます。
![スクリーンショット 2023-12-01 15.42.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/74c592b7-0959-f5f0-5d84-d2b72cfbf830.png)
- Processorに**Grok**を選択
- Fieldに**message**と入力
- Patternsに以下を入力
    - `%{TIMESTAMP_ISO8601:@timestamp} %{HOSTNAME:host.name} %{WORD:process.name}\[%{NUMBER:process.pid:int}\]: conn=%{NUMBER:connection_id:int} %{GREEDYDATA:rest}`
    - `%{}`の中身で右側(`@timestamp`とか)がElasticsearchでのフィールド名になり、`@timestamp`に時間が格納される
    - 共通項以外は一旦`rest`というフィールドに収める

![スクリーンショット 2023-12-01 15.47.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/9dadb203-2fed-94ba-643c-30c70a2ceb8e.png)
## 2つ目のGrok
ここから**Add a processor**をクリックして直列に処理を繋いでいきます。
次は`fd`か`op`で分かれますので、複数処理を書いていきます。
![スクリーンショット 2023-12-02 17.31.13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/b26cdb78-58c4-81b3-f2ba-1d88004e71ad.png)

- 処理するフィールドには`rest`を入力します
- 先程最後に残した`rest`に対して処理を行います。
    - `fd=%{NUMBER:fd:int} %{WORD:action} from IP=%{IP:source.ip}:%{NUMBER:source.port:int} \(IP=%{IP:destination.ip}:%{NUMBER:destination.port:int}\)`
        - 最初は`fd=12 ACCEPT from IP=192.168.48.3:54620 (IP=0.0.0.0:389)`にマッチするように記述
        - これにマッチしなかったら次のパターンが処理される
    - `op=%{WORD:operation} %{WORD:action} %{GREEDYDATA:rest}`
        - opから始まるものは4種類あるので一旦`rest`に格納
    - `fd=%{NUMBER:fd:int} %{WORD:action}`
        - `fd=12 closed`にマッチ
    - `op=%{WORD:operation} %{WORD:action}`
        - 上のGREEDYDATAの部分には`op=1 UNBIND`がマッチしないのでこれを追加

## 3つ目のGrok
次に`op`から始まるものを切っていきます。
![スクリーンショット 2023-12-01 16.00.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/bb8da651-060f-db9a-6449-6513a42bd71c.png)
- 更に`rest`に対して処理を行います
    - `dn="uid=%{DATA:uid},ou=%{DATA:organizational_unit},dc=%{DATA:domain_component},dc=%{DATA:top_level_domain}" %{GREEDYDATA:rest}`
    - `tag=%{NUMBER:tag:int} err=%{NUMBER:err:int} text=%{DATA:result_text}`
- このプロセスからここでは全くマッチしないパターンが出てきますので(fdが入っているもの)、エラーで止まらないように一番下にある**Ignore failures for this processor**をチェックしておきます
![スクリーンショット 2023-12-01 16.13.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/b82482e4-aa02-4f12-000e-6dd3db7d1331.png)

## 4つ目のGrok
最後に`dn=`から始まる2パターンについて区切ります。
![スクリーンショット 2023-12-02 14.24.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/1a273de4-ae6d-7a18-cf11-428d0d815cfc.png)
- `method=%{NUMBER:method:int}`
- `mech=%{DATA:mech} ssf=%{NUMBER:ssf:int}`

これで全パターンを区切ることが出来ます。
## Remove
最後に`rest`フィールドは必要ないので消しておきます。
![スクリーンショット 2023-12-02 14.27.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/119171c7-f86a-1b1b-7ee6-22fb81d3b049.png)
## On Failure
一応失敗したときの処理も加えておきます。
![スクリーンショット 2023-12-02 14.28.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/9d02c47a-22fa-9851-ce2a-aa3367cd59a8.png)
**Failure procesors**の**Add a processor**をクリックして、以下のように入力します。
![スクリーンショット 2023-12-02 14.30.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/a46c16d1-ed5b-adb2-163f-d0b347baf759.png)
Valueに記載している`{{ _ingest.on_failure_message }}`はエラーの内容です。

## 最終形
最終的にこうなっていると思います。
![スクリーンショット 2023-12-02 14.33.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/7853be79-f59b-be30-6a7f-1e1af2b1904f.png)
**Save pipeline**を最後にクリックするのを忘れないようにしましょう。

## いちいち入力が面倒くさい方へ
`pipeline.json`の中身をコピーして、Management -> Dev Toolsを開きます。
コピーした内容を左側に貼り付けて、右上の矢印をクリックすると上記の設定が全て行われます。
![スクリーンショット 2023-12-02 17.41.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/efe8138d-a4df-bdf9-da1d-20007d31a5ef.png)

# IntegrationsでIngest Pipelineの使用を設定
先ほど作成したIngest Pipelineは作っただけでは勝手には使用されません。Elastic AgentのCustom Logs Integrationsで設定します。
Management -> FleetでElastic Agentをインストールしたホストの横にあるAgent Policyを選びます。（ここでは**Agent Policy 1**)
![スクリーンショット 2023-12-02 14.48.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/593b472d-e71d-7240-043e-27ee60c2a200.png)
以下の画面に移るので、Custom LogsのName(ここでは**log-1**)をクリックします。
![スクリーンショット 2023-12-02 14.50.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/2613e76e-a2ab-bbeb-2202-dc065bf8472f.png)
先程設定したCustom Logsの設定画面が開きますので、**Change defaults**、**Advanced options**とクリックして、設定画面を開いていきます。
下記の**Custom configuration**に`pipeline: ldap`と入力し、右下の**Save integrations**をクリックして保存します。
![スクリーンショット 2023-12-02 15.04.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/8d06b6b4-3cea-428b-e6b3-02ad1fc8a011.png)

これでingest pipelineで行われる処理が反映されているはずです。
この手順は以下に記載されています。
https://www.elastic.co/guide/en/elasticsearch/reference/master/ingest.html#pipeline-custom-logs-configuration

# Discoverで確認
ではDiscoverでパースされているかどうか確認してみましょう。
`uid`が入っているログを選んでみてみましょう。
![スクリーンショット 2023-12-02 15.07.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/6275d122-94c2-6d8d-66bd-d6f254f8a480.png)
確かに`uid`がちゃんとフィールドとして入っています。

では、左側のフィールド一覧を一番下までスクロールして`uid`をクリックしてみます。
![スクリーンショット 2023-12-02 15.10.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/f3f09b63-7f70-6d68-097e-9c5286193869.png)
右側に`uid`の値の分布が表示されます。下のVisualizeをクリックしてみましょう。
![スクリーンショット 2023-12-02 15.12.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/beefb4ce-fa6a-12fc-98f0-f861e1209b07.png)
これが**Lens**というKibanaの可視化ツールです。ここで先程パースしたいろいろなデータを可視化してみてください。

# まとめ
よく使われるソフトやFirewall機器等は元々Integrationsが用意されているため、このような手順は必要ありません。
ただ独自のアプリケーションなどのログを取り込む場合はどうしてもこのような作業が必要となりますが、一度パースしてしまえば、Kibanaの強力な可視化ツールでデータを分析することが可能となります。
是非とも挑戦してみてください。

# おまけ：Grokについて
Grokパターンはあっているかどうか確かめながら行うことになると思います。KibanaにはGrokデバッガというそのためのツールを用意してあります。
Management -> Dev Tools -> Grok Debuggerで以下の画面が開きます。
![スクリーンショット 2023-12-02 15.20.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603229/9a771dde-75c6-82b8-206f-0a64b6c534e6.png)
**Sample Data**にログを入れて、**Grok Pattern**に試してみたいパターンを入れて、**Simulate**をクリックすると、Elasticsearchでどのように区切られるかが表示されます。

今回のリポジトリを使用してもらうと、Flaskのログも同じ`log`フォルダに`flask.log`として出力されています。
```
127.0.0.1 - - [02/Dec/2023 06:02:04] "GET /login?next=/ HTTP/1.1" 200 -
127.0.0.1 - - [02/Dec/2023 06:02:04] "POST /login HTTP/1.1" 200 -
```
これのパースにも是非チャレンジしてみてください。