Elasticsearchにカスタム時系列データを取り込む (Filebeat編)

先日の [Elasticsearchにカスタム時系列データを取り込む(Elastic Agent編)](https://qiita.com/yukshimizu/items/abec12ac46749db46b40) では、Elatic Agentに焦点を当てて、カスタムな時系列データの取り込み方法を解説しましたが、今回は、同じことをBeats (実際にはFilebeat) でやってみたいと思います。

## Integrationsって？
ちょっと前まで、[Elastic Stack](https://www.elastic.co/jp/elastic-stack/) と言えば、Elasticsearch、Kibana、Logstash、Beatsということになっていたのですが、最新のElastic Stackを見ると、LogstashとBeatsのところが、[Integrations(日本語だと「統合機能」)](https://www.elastic.co/jp/integrations)という風に変わっています。でも、このIntegrationsというのは、幅広いコネクター的なもの全般を指していて、お馴染みのLogstashやBeatsも、ちゃんと現役ですので、ご安心ください。

Kibanaの **Integrations > Browse integrations** からIntegrationsの一覧を見ることが出来ますが、`Beats only`をチェックすると、いわゆるBeats Modulesが表示されます。
![Beats_Integrations.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603118/81645b25-d7ac-cb20-6e2f-a23c3bd922a1.png)

### Beats
[Beats](https://www.elastic.co/jp/beats/) については、ご存知の方も多いと思いますが、軽量のデータシッパーとして様々な種類のBeatが存在します。[Elastic Agent](https://www.elastic.co/jp/elastic-agent) がリリースされて以来、少し影が薄くなった感があるのですが、以前のブログでも触れた通り、実際にはElastic Agentのコントロールの下、データプレーンとして動いています。また、一部のBeat設定については、まだElastic Agent経由で変更出来なかったりするので、現実にはそうしたケースにおいては、各種Beatを直接使うことも沢山あります。

### Filebeat
今回は [Filebeat](https://www.elastic.co/jp/beats/filebeat) を使って、カスタムの時系列データをElasticsearchに取り込んでみます。Filebeatは、言わずと知れた軽量のログシッパーですが、前述の [Filebeat Modules](https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-modules.html) によるコネクター群で様々なデータソースに対応していますし、カスタムなデータソースについても、[Filebeat Processors](https://www.elastic.co/guide/en/beats/filebeat/current/filtering-and-enhancing-data.html)を使って、ETL処理を行うことも出来ます。

## 実際にやってみよう
さて [前回](https://qiita.com/yukshimizu/items/abec12ac46749db46b40) は、Elastic Agentでデータを取得して、データのパース処理は、主にElasticsearchの [Ingest Pipelines](https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html) を使いました。ただし、Filebeat + Ingest Pipelineだと、ほとんど前回と同じになってしまうので、今回はエージェントサイド、つまりFilebeat Processorsを使って、Filebeatだけで同じことを実現してみたいと思います。

### 1. NewsAPI
前回同様、[NewsAPI](https://newsapi.org/)という、世の中のニュースのヘッドラインをJSON形式で返してくれるAPIをデータソースとして使います。登録して、API Keyを取得します。パラメータによって、対象を絞り込むことができるのですが、[Japanのヘッドラインを取得する](https://newsapi.org/s/japan-news-api)とこんな感じで返ってきます。
```
GET https://newsapi.org/v2/top-headlines?country=jp&apiKey=hogehoge
```
レスポンス:
```json
{
  "status": "ok",
  "totalResults": 30,
  "articles": [
    {
        "source": {
          "id": null,
          "name": "News.local"
        },
        "author": "ローカルニュース",
        "title": "これはグッドニュースだ！",
        "description": "今日、素晴らしい出来事がありました！",
        "url": "https://news.local/hogehoge.html",
        "urlToImage": "https://news.local/hogehoge/hogehoge.jpg",
        "publishedAt": "2022-05-25T05:49:58Z",
        "content": null
    },
    {
      (中略)
    }
  ]
}
```
### 2. Inputの設定
Filebeatの設定は、基本的には`filebeat.yml`で行います。今回はNewsAPIからのInputということになるので、[HTTP JSON input](https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-input-httpjson.html) を使います。NewsAPIから返されるJSONドキュメントを見ると、必要なデータは、`articles`配列の要素であることがわかるので、[Response Split](https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-input-httpjson.html#response-split) の機能を使って、配列の要素毎にイベントを生成するようにします。これで前述の例のように、30のニューストピックスが配列として返された場合は、30の独立したイベントが生成されるようになります。ここは前回と基本は同じです。Outputはとりあえず、デバッグ用にコンソールに出力すればよいので、[Console output](https://www.elastic.co/guide/en/beats/filebeat/current/console-output.html) を指定して、デフォルトの [Elasticsearch output](https://www.elastic.co/guide/en/beats/filebeat/current/elasticsearch-output.html) はコメントアウトしておきます(Filebeatでは同時に複数のoutputは書けません)。

```yml:filebeat.yml
# ============================== Filebeat inputs ===============================
# News API Configuration

filebeat.inputs:
- type: httpjson
  interval: 1m
  request.url: https://newsapi.org/v2/top-headlines?country=jp&apiKey=hogehoge
  response.split:
    target: body.articles
    type: array

# ================================== Outputs ===================================
# Configure what output to use when sending the data collected by the beat.

output.console:
  pretty: true

# ---------------------------- Elasticsearch Output ----------------------------
# output.elasticsearch:
  # Array of hosts to connect to.
#   hosts: ["localhost:9200"]
```

この段階で、Filebeatからの出力は以下のようになります。
```json
{
  "@timestamp": "2022-05-26T02:38:00.173Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "8.2.1"
  },
  "event": {
    "created": "2022-05-26T02:38:00.173Z"
  },
  "message": "{\"author\":\"ローカルニュース\",\"content\":null,\"description\":\"今日、素晴らしい出来事がありました！\",\"publishedAt\":\"2022-05-25T05:49:58Z\",\"source\":{\"id\":null,\"name\":\"News.local\"},\"title\":\"これはグッドニュースだ！",\"url\":\"https://news.local/hogehoge.html\",\"urlToImage\":\"https://news.local/hogehoge/hogehoge.jpg\"}",
  "input": {
    "type": "httpjson"
  },
  "ecs": {
    "version": "8.0.0"
  },
  "host": {
    "name": "hogehoge.local",
    "id": "22570515-B9D2-5E7A-BD39-608FA0BB373C",
    "ip": [
      (中略)
    ],
    "mac": [
      (中略)
    ],
    "hostname": "hogehoge.local",
    "architecture": "x86_64",
    "os": {
      "family": "darwin",
      "name": "Mac OS X",
      "kernel": "19.6.0",
      "build": "19H1922",
      "type": "macos",
      "platform": "darwin",
      "version": "10.15.7"
    }
  },
  "agent": {
    "version": "8.2.1",
    "ephemeral_id": "6506efe3-93ad-4e10-b793-701005f33fdf",
    "id": "2000b9bc-a219-438f-bdda-4e6d64febaca",
    "name": "hogehoge.local",
    "type": "filebeat"
  }
}
```

### 3. JSON文字列をデコードする
次に`meeeage`フィールドのJSON文字列を、`news`フィールドにデコードします。前回はIngest Pipelineで対応しましたが、今回は、Filebeatの [Decode JSON fields](https://www.elastic.co/guide/en/beats/filebeat/current/decode-json-fields.html) processorを使います。`processors`の定義の下に、以下のように追加します。元の`message`フィールドは不要なので、[Drop fields](https://www.elastic.co/guide/en/beats/filebeat/current/drop-fields.html)で削除しておきます。

```yml:filebeat.yml
# ================================= Processors =================================
processors:
  (中略)
# Adding configurations for News API
  - decode_json_fields:
      fields: ["message"]
      target: "news"
  - drop_fields:
      fields: ["message"]
```
そうすると、出力は以下のようになります。
```json
{
  "@timestamp": "2022-05-26T02:40:33.341Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "8.2.1"
  },
  "input": {
    "type": "httpjson"
  },
  "ecs": {
    "version": "8.0.0"
  },
  "host": {
    "os": {
      "version": "10.15.7",
      "family": "darwin",
      "name": "Mac OS X",
      "kernel": "19.6.0",
      "build": "19H1922",
      "type": "macos",
      "platform": "darwin"
    },
    "id": "22570515-B9D2-5E7A-BD39-608FA0BB373C",
    "ip": [
      (中略)
    ],
    "name": "hogehoge.local",
    "mac": [
      (中略)
    ],
    "hostname": "hogehoge.local",
    "architecture": "x86_64"
  },
  "agent": {
    "ephemeral_id": "bd131899-cfa9-4f1f-b5d5-48b387c32171",
    "id": "2000b9bc-a219-438f-bdda-4e6d64febaca",
    "name": "hogehoge.local",
    "type": "filebeat",
    "version": "8.2.1"
  },
  "news": {
    "description": "今日、素晴らしい出来事がありました！",
    "publishedAt": "2022-05-25T05:49:58Z",
    "source": {
      "id": null,
      "name": "News.local"
    },
    "title": "これはグッドニュースだ！",
    "url": "https://news.local/hogehoge.html",
    "urlToImage": "https://news.local/hogehoge/hogehoge.jpg",
    "author": "ローカルニュース",
    "content": null
  },
  "event": {
    "created": "2022-05-26T02:40:33.341Z"
  }
}
```
意図した通りに`news`フィールドにデコードされました。
### 4. null valueフィールドに対応する
さらに、前回同様`source.id`や`author`がnullである`article`の場合は、`source.name`をセットするようにしたいと思います。[FilebeatのProcessorには条件を書ける](https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html)ので、次のように条件を追加して、[Copy fields](https://www.elastic.co/guide/en/beats/filebeat/current/copy-fields.html) processorを使って、フィールドをコピーするようにしてみます。

```yml
- copy_fields:
    when:
      equals:
        news.source.id: null
    fields:
      - from: news.source.name
        to: news.source.id
```
ところが、実行すると以下のようにエラーとなってしまいます。結論から言うと、フィールドの値が`null`かどうかというチェックは出来ません。というのも、条件には [integerかstringの値しか書けない](https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#condition-equals)からです。しかも、NewsAPIから返されるJSONでは、明確に`null`となっており、空文字`""`では判定できません。
```
$ ./filebeat
Exiting: error initializing processors: failed to initialize condition: condition attempted to set 'news.source.id' -> '<nil>' and encountered unexpected type '<nil>', only strings, ints, and booleans are allowed
```

困ったなと思いきや、実はFilebeat Processorには、[Script](https://www.elastic.co/guide/en/beats/filebeat/current/processor-script.html) processorという万能ナイフがあるのです。一言でいうと、JavaScriptが書けちゃいます。ここでは、`source.id`と`author`それぞれが`null`だったら、一旦フィールドをDeleteし、その上でフィールドが存在していなければ、`source.name`をコピーするように設定します。

```yml:filebeat.yml
# Adding configurations for News API
  - decode_json_fields:
      fields: ["message"]
      target: "news"
  - drop_fields:
      fields: ["message"]
  - script:
      lang: javascript
      source: >
        function process(event) {
            if (event.Get("news.source.id") == null) {
                event.Delete("news.source.id");
            }
            if (event.Get("news.author") == null) {
                event.Delete("news.author");
            }
            return event;
        }
  - copy_fields:
      when.not.has_fields: ["news.source.id"]
      fields:
        - from: news.source.name
          to: news.source.id
  - copy_fields:
      when.not.has_fields: ["news.author"]
      fields:
        - from: news.source.name
          to: news.author
```

### 5. @&#xFFF0;timestampを書き換える
仕上げに、イベントの`@timestamp`を、処理した時間である**processing time**から、イベントが発生した時間である**event time**に書き換えます。これは、[Timestamp](https://www.elastic.co/guide/en/beats/filebeat/current/processor-timestamp.html) processorで処理することができます。`puhlishedAt`が、ニュース記事が投稿された時間なので、このフィールドを使います。`layouts`で指定するフォーマットの時刻は、[GOのtimeパッケージ特有](https://pkg.go.dev/time#pkg-constants)のもので、この時刻でなくてはなりません。

```yml:filebeat.yml
# Adding configurations for News API
  - decode_json_fields:
      fields: ["message"]
      target: "news"
  - drop_fields:
      fields: ["message"]
  - script:
      lang: javascript
      source: >
        function process(event) {
            if (event.Get("news.source.id") == null) {
                event.Delete("news.source.id");
            }
            if (event.Get("news.author") == null) {
                event.Delete("news.author");
            }
            return event;
        }
  - copy_fields:
      when.not.has_fields: ["news.source.id"]
      fields:
        - from: news.source.name
          to: news.source.id
  - copy_fields:
      when.not.has_fields: ["news.author"]
      fields:
        - from: news.source.name
          to: news.author
  - timestamp:
      field: news.publishedAt
      layouts:
        - '2006-01-02T15:04:05Z'
```
コンソールへの出力は、以下のようになります。
```json
{
  "@timestamp": "2022-05-25T05:49:58.000Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "8.2.1"
  },
  "agent": {
    "version": "8.2.1",
    "ephemeral_id": "52c51fef-d0f3-4655-aaec-d5a15f81cc71",
    "id": "2000b9bc-a219-438f-bdda-4e6d64febaca",
    "name": "hogehoge.local",
    "type": "filebeat"
  },
  "ecs": {
    "version": "8.0.0"
  },
  "news": {
    "description": "今日、素晴らしい出来事がありました！",
    "publishedAt": "2022-05-25T05:49:58Z",
    "source": {
      "id": "News.local",
      "name": "News.local"
    },
    "title": "これはグッドニュースだ！",
    "url": "https://news.local/hogehoge.html",
    "urlToImage": "https://news.local/hogehoge/hogehoge.jpg",
    "author": "ローカルニュース",
    "content": null
  },
  "event": {
    "created": "2022-05-26T02:49:11.475Z"
  },
  "input": {
    "type": "httpjson"
  },
  "host": {
    "hostname": "hogehoge.local",
    "architecture": "x86_64",
    "os": {
      "version": "10.15.7",
      "family": "darwin",
      "name": "Mac OS X",
      "kernel": "19.6.0",
      "build": "19H1922",
      "type": "macos",
      "platform": "darwin"
    },
    "id": "22570515-B9D2-5E7A-BD39-608FA0BB373C",
    "ip": [
      (中略)
    ],
    "mac": [
      (中略)
    ],
    "name": "hogehoge.local"
  }
}
```

### 6. Index Templateを修正する
ここまでは、デバッグしやすいようにコンソールに出力していただけですが、最後にElasticsearchに投入していきたいと思います。Elasticsearchにこのまま投入すると、`news`フィールドはDynamic Mapingで作成されてしまうため、配下の全てのフィールドは`keyword`になってしまいます。きちんとIndexのMappingを定義するために、[Index Template](https://www.elastic.co/guide/en/elasticsearch/reference/current/index-templates.html) に反映したいところです。デフォルトのFilebeatのIndex Templateは、8系では`filebeat-8.x.x`になりますが、試しにKibanaのDev Toolsで確認すると、あまりの巨大さに驚愕します。
```
GET _index_template/filebeat-8.2.1
```
レスポンス:
![filebeat-8.2.1_template.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603118/53f40ec2-b04c-bb03-53d5-17779aff8fd8.png)

優に3万ラインを超えてます。Beats系では、このように1つのIndex Templateで対象のIndex(この場合、filebeat-8.x.x)の全てのMappingを管理する形になるので、Beats Modulesの拡大によるフィールドの増加に伴って、徐々に管理性が低くなっていました。そこで、[Elastic Agentでは、Dataset毎に出力するData streamを分けて、それぞれがIndex Templateを持つようになりました。](https://www.elastic.co/guide/en/fleet/current/data-streams.html)ちなみに、Beatsの8系では、従来のIndexへの出力から、Elastic Agentと同様に [Data stream](https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html) への出力に変わっています。

さて、やりたいことは、以下の`news`フィールドのMappingを追加したいだけなのですが、そのために先ほどの`filebeat-8.x.x` Index Templateを直接修正するのは気が引けますし、修正したらバージョンアップ時のメンテナンスも面倒です。
```json
{
  "news" : {
    "properties" : {
      "author" : {
        "ignore_above" : 1024,
        "type" : "keyword",
        "fields" : {
          "text" : {
            "type" : "text"
          }
        }
      },
      "content" : {
        "ignore_above" : 1024,
        "type" : "keyword",
        "fields" : {
          "text" : {
            "type" : "text"
          }
        }
      },
      "description" : {
        "ignore_above" : 1024,
        "type" : "keyword",
        "fields" : {
          "text" : {
            "type" : "text"
          }
        }
      },
      "publishedAt" : {
        "type" : "date"
      },
      "source" : {
        "properties" : {
          "id" : {
            "type" : "keyword",
            "ignore_above" : 1024
          },
          "name" : {
            "ignore_above" : 1024,
            "type" : "keyword",
            "fields" : {
              "text" : {
                "type" : "text"
              }
            }
          }
        }
      },
      "title" : {
        "ignore_above" : 1024,
        "type" : "keyword",
        "fields" : {
          "text" : {
            "type" : "text"
          }
        }
      },
      "url" : {
        "type" : "keyword",
        "ignore_above" : 1024
      },
      "urlToImage" : {
        "type" : "keyword",
        "ignore_above" : 1024
      }
    }
  }
}
```
こんな時に使える設定が、`setup.template.append_fields`です。詳細は[公式ドキュメント](https://www.elastic.co/guide/en/beats/filebeat/current/configuration-template.html#configuration-template)を参照頂ければと思いますが、`setup.template.overwrite`と合わせ技で、`filebeat.yml`上でIndex Templateに追加したいフィールドを定義することが出来ます(頭をJSONからYAMLに切り替える必要がありますが...)。

```yml:filebeat.yml
# Adding news field
setup.template.overwrite: true
setup.template.append_fields:
- name: news.author
  type: keyword
  ignore_above: 1024
  multi_fields:
    - type: text
      norms: true
      name: text
- name: news.content
  type: keyword
  ignore_above: 1024
  multi_fields:
    - type: text
      norms: true
      name: text
- name: news.description
  type: keyword
  ignore_above: 1024
  multi_fields:
    - type: text
      norms: true
      name: text
- name: news.publishedAt
  type: date
- name: news.source.id
  type: keyword
  ignore_above: 1024
- name: news.source.name
  type: keyword
  ignore_above: 1024
  multi_fields:
    - type: text
      norms: true
      name: text
- name: news.title
  type: keyword
  ignore_above: 1024
  multi_fields:
    - type: text
      norms: true
      name: text
  - name: news.url
  type: keyword
  ignore_above: 1024
  - name: news.urlToImage
  type: keyword
  ignore_above: 1024

```
Multi-fieldsも書けますし、`filebeat.yml`で一元管理も出来るのでシンプルです。注意点としては、`text`フィールドの時に、`norms`を明示的に指定しないと`norms: false`で作成されることです。Elasticsearchにおいては、`norms`はデフォルト`true`なのですが...

あとは、今回のようにfilebeatのプロセスが1つの場合は特に問題になりませんが、複数の場合、全てのfilebeatプロセスがIndex Templateを上書きするので、差分があったりすると後勝ちになりますので、注意が必要です。

### 7. Elasticsearchに投入する
いよいよ、Elasticsearchへの投入です。`filebeat.yml`で、outputを変更します。[Elastic Cloudを使っている場合](https://www.elastic.co/guide/en/beats/filebeat/current/configure-cloud-id.html)は、次のように`cloud.id`と`cloud.auth`を書いてあげるだけでOKです。

```yml:filebeat.yml
# =============================== Elastic Cloud ================================

# These settings simplify using Filebeat with the Elastic Cloud (https://cloud.elastic.co/).

# The cloud.id setting overwrites the `output.elasticsearch.hosts` and
# `setup.kibana.host` options.
# You can find the `cloud.id` in the Elastic Cloud web UI.
cloud.id: "staging:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyRjZWM2ZjI2MWE3NGJmMjRjZTMzYmI4ODExYjg0Mjk0ZiRjNmMyY2E2ZDA0MjI0OWFmMGNjN2Q3YTllOTYyNTc0Mw=="

# The cloud.auth setting overwrites the `output.elasticsearch.username` and
# `output.elasticsearch.password` settings. The format is `<user>:<pass>`.
cloud.auth: "elastic:YOUR_PASSWORD"
```
オンプレでElasticsearchクラスタを立てている場合は、公式ドキュメントの [Elasticsearch output](https://www.elastic.co/guide/en/beats/filebeat/current/elasticsearch-output.html) を参考にしてみて下さい。

最終的には、こんな感じでIndexされます。
```json
{
  "_index": ".ds-filebeat-8.2.1-2022.05.27-000001",
  "_id": "OOwpBIEBCJsKRn8hWsLn",
  "_version": 1,
  "_score": 1,
  "_source": {
    "@timestamp": "2022-05-25T05:49:58Z.000Z",
    "input": {
      "type": "httpjson"
    },
    "host": {
      "name": "hogehoge.local",
      "architecture": "x86_64",
      "os": {
        "version": "10.15.7",
        "family": "darwin",
        "name": "Mac OS X",
        "kernel": "19.6.0",
        "build": "19H1922",
        "type": "macos",
        "platform": "darwin"
      },
      "id": "22570515-B9D2-5E7A-BD39-608FA0BB373C",
      "ip": [
        (中略)
      ],
      "mac": [
        (中略)
      ],
      "hostname": "hogehoge.local"
    },
    "agent": {
      "id": "2000b9bc-a219-438f-bdda-4e6d64febaca",
      "name": "hogehoge.local",
      "type": "filebeat",
      "version": "8.2.1",
      "ephemeral_id": "752dedb3-fd8c-4772-8ae2-38f8c3d61b92"
    },
    "ecs": {
      "version": "8.0.0"
    },
    "news": {
      "description": "今日、素晴らしい出来事がありました！",
      "publishedAt": "2022-05-25T05:49:58Z",
      "source": {
        "id": "News.local",
        "name": "News.local"
      },
      "title": "これはグッドニュースだ！",
      "url": "https://news.local/hogehoge.html",
      "urlToImage": "https://news.local/hogehoge/hogehoge.jpg",
      "author": "ローカルニュース",
      "content": null
    },
    "event": {
      "created": "2022-05-27T06:17:02.453Z"
    }
  },
}
```

## まとめ
今回は、[前回](https://qiita.com/yukshimizu/items/abec12ac46749db46b40) との比較の意味で、Filebeatだけで頑張ってみました。

| やりたいこと | Elastic Agent編 | Filebeat編 |
|:-:|:-:|:-:|
|HTTP APIでJSONを取り込む|Custom HTTPJSON Input (Elastic Agent)|HTTP JSON Input (Filebeat)|
|JSON文字列をデコード|JSON processor (Ingest Pipeline)|Decode JSON fields processor (Filebeat)|
|null-valueフィールド対応|Set processor (Ingest Pipeline)|Script processor (Filebeat)|
|@&#xFFF0;timestamp書き換え|Date processor (Ingest Pipeline)|Timestamp processor (Filebeat)|

[Filebeat Processos](https://www.elastic.co/guide/en/beats/filebeat/current/filtering-and-enhancing-data.html) だけでも、それなりのことが出来ることはご理解いただけたかと思いますが、現実の選択肢としては、エージェントサイドだけに処理を寄せるというのはレアケースかもしれません。こうしたパース処理をどこで実行するかは、言うまでもなくユースケースに依存するので、実際にはパフォーマンスのテストなどを通じて、最適な方法を探ることになります。Elasticでは [様々な選択肢](https://www.elastic.co/jp/integrations) があるので、色々と試してみてください。