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

これまで、[Elasticsearchにカスタム時系列データを取り込む(Elastic Agent編)](https://qiita.com/yukshimizu/items/abec12ac46749db46b40) 、及び [Elasticsearchにカスタム時系列データを取り込む(Filebeat編)](https://qiita.com/yukshimizu/items/eef4dffa435802567a44) で、Elastic AgentとFilebeatを使ったカスタムな時系列データの取り込みについて解説してきましたが、ここまで来たらということで、今回は、Logstashについて解説してみたいと思います。

## Logstash
言わずと知れた [Logstash](https://www.elastic.co/jp/logstash/) ですが、古くは [ELK Stack](https://www.elastic.co/jp/what-is/elk-stack) の**L**はLogstashに由来しています。
Logstashは、サーバーサイドの汎用のETLツールで、様々なデータの取り込みで使っている方も多いと思います。Logstashの特徴として、豊富なInput Pluginだけではなく、Outputの方もElasticsearchに限らず、S3やKafkaといった様々なデータストアに出力できるところが、汎用と言われている所以です。[Integrations(日本語だと「統合機能」)](https://www.elastic.co/jp/integrations)が出てきたことで、存在感が薄れてる？かもしれませんが、バッチリ現役ですので、ご安心ください。8系では、[Elastic Common Schema(ECS)](https://www.elastic.co/guide/en/ecs/current/ecs-reference.html) の互換性モードがデフォルトとなり、ECSへの対応も進んでいます。

## 実際にやってみよう
過去のブログでは、次のような方法で実現してきました。今回は、同じことをLogstashのPluginを使って実現してみたいと思います。

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

### 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-07-24T04:55:57Z",
        "content": null
    },
    {
      (中略)
    }
  ]
}
```
### 2.Inputの設定
Logstashの処理は、一般的にはPipelineと言われ、inputs → filters → outputs の3ステージで構成され、それぞれのステージでPluginの設定を行います。今回は、設定ファイルを`logstash-news.conf`として作成します。先ずはInputの設定になりますが、httpを介してNewsAPIからデータを所得するので、[Http_poller input plugin](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-http_poller.html) を使います。取得したデータは、`news`フィールドに格納しておきます。Outputはとりあえず、デバッグ用にコンソールに出力すればよいので、[Stdout output plugin](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-stdout.html) を設定しておきます。

```
input {
  http_poller {
    urls => {
      news => "https://newsapi.org/v2/top-headlines?country=jp&apiKey=hogehoge"
    }
    request_timeout => 60
    # Supports "cron", "every", "at" and "in" schedules by rufus scheduler
    schedule => { "every" => "1m" }
    target => "news"
    ecs_compatibility => "v1"
    ssl_verification_mode => "none"
  }
}

output {
  stdout {
    codec => rubydebug
  }
}
```
この段階での、Logstashからの出力は以下のようになります。
```
{
      "@version" => "1",
          "news" => {
            "@version" => "1",
               "event" => {
            "original" => "{\"status\":\"ok\",\"totalResults\":30,\"articles\":[(中略)]}"
        },
            "articles" => [
            [ 0] {
                "description" => "今日、素晴らしい出来事がありました！",
                     "author" => "ローカルニュース",
                        "url" => "https://news.local/hogehoge.html",
                    "content" => nil,
                     "source" => {
                      "id" => nil,
                    "name" => "News.local"
                },
                      "title" => "これはグッドニュースだ！",
                 "urlToImage" => "https://news.local/hogehoge/hogehoge.jpg",
                "publishedAt" => "2022-07-24T04:55:57Z"
            },
　　　　         (中略)
        ],
        "totalResults" => 30,
              "status" => "ok",
          "@timestamp" => 2022-07-24T07:09:51.828241Z
    },
    "@timestamp" => 2022-07-24T07:09:51.835379Z
}
```

### 3.配列をイベントに分割する
現状ですと、`news.event.original`に取得したデータがJSON文字列として格納され、さらに`articles`配列の要素にそれぞれの記事がデコードされていることがわかりますが、これは、Http_poller input pluginのデフォルトのcodecが`json`になっているからで、[Json codec plugin](https://www.elastic.co/guide/en/logstash/current/plugins-codecs-json.html) のおかげで暗黙的にデコードされているからです。いずれにせよ1つのイベントとして出力されてしまっていて、これは以前のブログの、Elastic AgentやFilebeatの時と少し異なる挙動ですね。配列要素を、それぞれ独立したイベントとして出力するようにしてあげる必要があります。ここで、[Split filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-split.html) の出番です。以下の構成では、対象となる`news.articles`配列を`news`フィールドにそれぞれ分割すると同時に、元々のJSON文字列が格納されている`news.event`フィールドは不要となるので、削除するようにしています。ちなみにネストされたフィールドへのアクセスは、`[top-level field][nested field]`という書き方になるので、少し注意が必要です。この辺りの書き方は、[公式ドキュメント](https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#logstash-config-field-references)を参照してみてください。
```
(前略)

filter {
  split {
    field => "[news][articles]"
    target => "news"
    remove_field => "[news][event]"
  }
}

(後略)
```
そうすると、配列要素がそれぞれのイベントとして出力されるようになります。
```
 {
    "@timestamp" => 2022-07-24T07:40:02.507870Z,
          "news" => {
        "description" => "今日、素晴らしい出来事がありました！",
         "urlToImage" => "https://news.local/hogehoge/hogehoge.jpg",
        "publishedAt" => "2022-07-24T04:55:57Z",
                "url" => "https://news.local/hogehoge.html",
             "source" => {
              "id" => nil,
            "name" => "News.local"
        },
             "author" => nil,
              "title" => "これはグッドニュースだ！",
            "content" => nil
    },
      "@version" => "1"
}
 ```
### 4.null valueフィールドに対応する
さらに、前回同様`source.id`や`author`がnullである`article`の場合は、`source.name`をセットするようにしたいと思います。ここでは、[Mutate filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html) を使います。Logstashの場合、[条件文は簡単に書ける](https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#conditionals)ので、違和感はないでしょう。
```
(前略)

filter {
  split {
    field => "[news][articles]"
    target => "news"
    remove_field => "[news][event]"
  }

  if ![news][author] {
    mutate {
      copy => { "[news][source][name]" => "[news][author]" }
    }
  }

  if ![news][source][id] {
    mutate {
      copy => { "[news][source][name]" => "[news][source][id]" }
    }
  }
}

(後略)
```

### 5.＠timestampを書き換える
仕上げに、イベントの`@timestamp`を、処理した時間である**processing time**から、イベントが発生した時間である**event time**に書き換えます。これは、[Date filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html) で処理することができます。`puhlishedAt`が、ニュース記事が投稿された時間なので、このフィールドを使います。`publishedAt`の形式は、ISO8601形式でサポートされている形式なので、`formats`は`ISO8601`と指定しておきます。
```
(前略)

filter {
  split {
    field => "[news][articles]"
    target => "news"
    remove_field => "[news][event]"
  }

  if ![news][author] {
    mutate {
      copy => { "[news][source][name]" => "[news][author]" }
    }
  }

  if ![news][source][id] {
    mutate {
      copy => { "[news][source][name]" => "[news][source][id]" }
    }
  }

  date {
    match => [ "[news][publishedAt]", "ISO8601" ]
  }
}

(後略)
```
最終的に、コンソールへの出力は以下のようになります。
```
{
   "@timestamp" => 2022-07-24T04:55:57Z,
         "news" => {
       "description" => "今日、素晴らしい出来事がありました！",
        "urlToImage" => "https://news.local/hogehoge/hogehoge.jpg",
       "publishedAt" => "2022-07-24T04:55:57Z",
               "url" => "https://news.local/hogehoge.html",
            "source" => {
             "id" => "News.local",
           "name" => "News.local"
       },
            "author" => "News.local",
             "title" => "これはグッドニュースだ！",
           "content" => nil
   },
     "@version" => "1"
}
```
### 6.Elasticsearchに投入する
さて、とりあえずいい感じになってきたので、Elasticsearchに投入してみましょう。ElasticsearchへのIndexは、[Elasticsearch output plugin](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html) を使います。Logstashでも、7.13から [Data stream](https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html) へ出力することができるようになったので、Elastic AgentやFilebeatの時と同様に、Data streamに出力してみましょう。宛先も、[Elastic Cloudを使っている場合](https://www.elastic.co/guide/en/logstash/current/connecting-to-cloud.html)は、次のように`cloud.id`と`cloud.auth`を書いてあげるだけでOKです。
```
(前略)

output {
  elasticsearch {
    cloud_id => "staging:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyRjZWM2ZjI2MWE3NGJmMjRjZTMzYmI4ODExYjg0Mjk0ZiRjNmMyY2E2ZDA0MjI0OWFmMGNjN2Q3YTllOTYyNTc0Mw=="
    cloud_auth => "elastic:YOUR_PASSWORD"
    data_stream => "true"
  }
}
```
そうすると、デフォルトの場合、`logs-generic-default`というData streamにIndexされることになります。[Index Template](https://www.elastic.co/guide/en/elasticsearch/reference/current/index-templates.html) もデフォルトの`logs`というテンプレートが適用されており、例によって`news`フィールドはDynamic Mappingで、配下の全てのフィールドが`keyword`になってしまっています。
```
GET logs-generic-default/_mapping
```
レスポンス:
```json
{
  ".ds-logs-generic-default-2022.07.23-000001": {
    "mappings": {
      "_data_stream_timestamp": {
        "enabled": true
      },
      "dynamic_templates": [
        {
          "match_ip": {
            "match": "ip",
            "match_mapping_type": "string",
            "mapping": {
              "type": "ip"
            }
          }
        },
        {
          "match_message": {
            "match": "message",
            "match_mapping_type": "string",
            "mapping": {
              "type": "match_only_text"
            }
          }
        },
        {
          "strings_as_keyword": {
            "match_mapping_type": "string",
            "mapping": {
              "ignore_above": 1024,
              "type": "keyword"
            }
          }
        }
      ],
      "date_detection": false,
      "properties": {
        "@timestamp": {
          "type": "date"
        },
        "@version": {
          "type": "keyword",
          "ignore_above": 1024
        },
        "data_stream": {
          "properties": {
            "dataset": {
              "type": "constant_keyword",
              "value": "generic"
            },
            "namespace": {
              "type": "constant_keyword",
              "value": "default"
            },
            "type": {
              "type": "constant_keyword",
              "value": "logs"
            }
          }
        },
        "ecs": {
          "properties": {
            "version": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "host": {
          "type": "object"
        },
        "news": {
          "properties": {
            "author": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "content": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "description": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "publishedAt": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "source": {
              "properties": {
                "id": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "name": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            },
            "title": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "url": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "urlToImage": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        }
      }
    }
  }
}
```
logsテンプレート
```
GET _index_template/logs/
```
レスポンス:
```json
{
  "index_templates": [
    {
      "name": "logs",
      "index_template": {
        "index_patterns": [
          "logs-*-*"
        ],
        "composed_of": [
          "logs-mappings",
          "data-streams-mappings",
          "logs-settings"
        ],
        "priority": 100,
        "version": 1,
        "_meta": {
          "description": "default logs template installed by x-pack",
          "managed": true
        },
        "data_stream": {
          "hidden": false,
          "allow_custom_routing": false
        },
        "allow_auto_create": true
      }
    }
  ]
}
```
### 7.Index Templateを作成する
ここでは、デフォルトの`logs-generic-default` Data streamの設定を生かしつつ、`news`フィールドのMappingを追加することにしたいところです。まず、`logs-logstash.news-qa`というData streamを使うことにするとして、Kibanaの **Stack Management > Index Management > Index Templates** から、`logs` Index Templateを、`logs-logstash.news`という名前でクローンします。その時、Index patternsには、`logs-logstash.news-*`と指定します。

![clone-template-logs1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603118/7c95361b-207e-9312-3c07-22855696567f.png)


さらに、カスタマイズ部分の [Component Template](https://www.elastic.co/guide/en/elasticsearch/reference/current/set-up-a-data-stream.html#create-component-templates) として、`logs-logsash.news@custom`を作成します。[ILM Policy](https://www.elastic.co/guide/en/elasticsearch/reference/current/index-lifecycle-management.html) `index.lifecycle.name`には、ビルトインの`logs`を設定しています。以下は、APIの入力ですが、もちろんKibanaのUIから作成してもOKです。
```json
PUT _component_template/logs-logstash.news@custom
{
  "template": {
    "settings": {
      "index": {
        "lifecycle": {
          "name": "logs"
        },
        "codec": "best_compression",
        "query": {
          "default_field": [
            "*"
          ]
        }
      }
    },
    "mappings": {
      "dynamic_templates": [],
      "properties": {
        "news": {
          "type": "object",
          "properties": {
            "publishedAt": {
              "type": "date"
            },
            "author": {
              "ignore_above": 1024,
              "type": "keyword",
              "fields": {
                "text": {
                  "type": "text"
                }
              }
            },
            "urlToImage": {
              "ignore_above": 1024,
              "type": "keyword"
            },
            "description": {
              "ignore_above": 1024,
              "type": "keyword",
              "fields": {
                "text": {
                  "type": "text"
                }
              }
            },
            "source": {
              "type": "object",
              "properties": {
                "name": {
                  "ignore_above": 1024,
                  "type": "keyword",
                  "fields": {
                    "text": {
                      "type": "text"
                    }
                  }
                },
                "id": {
                  "ignore_above": 1024,
                  "type": "keyword"
                }
              }
            },
            "title": {
              "ignore_above": 1024,
              "type": "keyword",
              "fields": {
                "text": {
                  "type": "text"
                }
              }
            },
            "content": {
              "ignore_above": 1024,
              "type": "keyword",
              "fields": {
                "text": {
                  "type": "text"
                }
              }
            },
            "url": {
              "ignore_above": 1024,
              "type": "keyword"
            }
          }
        }
      }
    }
  }
}
```
そして、`logs-logstash.news` Index Templateで、`logs-logsash.news@custom`を含むようにします。`logs-mappings`と`data-streams-mappings`は残しておきます。

![clone-template-logs2.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2603118/8422bef6-1a3a-565f-5c56-7bd7f8f9764c.png)


`logs-logstash.news` Index Templateは、以下のようになります。
```
GET _index_template/logs-logstash.news/
```
レスポンス:
```json
{
  "index_templates": [
    {
      "name": "logs-logstash.news",
      "index_template": {
        "index_patterns": [
          "logs-logstash.news-*"
        ],
        "composed_of": [
          "logs-mappings",
          "data-streams-mappings",
          "logs-logstash.news@custom"
        ],
        "priority": 249,
        "version": 1,
        "_meta": {
          "description": "logstash news template",
          "managed": false
        },
        "data_stream": {
          "hidden": false,
          "allow_custom_routing": false
        }
      }
    }
  ]
}
```

最後に、`logs-logstash.news-qa` Data streamに出力するように、Logstashの設定ファイルの`output`に反映します。最終的な`logstash-news.conf`は以下のようになります。
```
input {
  http_poller {
    urls => {
      news => "https://newsapi.org/v2/top-headlines?country=jp&apiKey=hogehoge"
    }
    request_timeout => 60
    # Supports "cron", "every", "at" and "in" schedules by rufus scheduler
    schedule => { "every" => "1m" }
    target => "news"
    ecs_compatibility => "v1"
    ssl_verification_mode => "none"
  }
}

filter {
  split {
    field => "[news][articles]"
    target => "news"
    remove_field => "[news][event]"
  }

  if ![news][author] {
    mutate {
      copy => { "[news][source][name]" => "[news][author]" }
    }
  }

  if ![news][source][id] {
    mutate {
      copy => { "[news][source][name]" => "[news][source][id]" }
    }
  }

  date {
    match => [ "[news][publishedAt]", "ISO8601" ]
  }
}

output {
  elasticsearch {
    cloud_id => "staging:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyRjZWM2ZjI2MWE3NGJmMjRjZTMzYmI4ODExYjg0Mjk0ZiRjNmMyY2E2ZDA0MjI0OWFmMGNjN2Q3YTllOTYyNTc0Mw=="
    cloud_auth => "elastic:YOUR_PASSWORD"
    data_stream => "true"
    data_stream_type => "logs"
    data_stream_dataset => "logstash.news"
    data_stream_namespace => "qa"
  }
}
```

最終的には、以下のようにIndexされます。
```json
{
  "_index": ".ds-logs-logstash.news-qa-2022.07.23-000001",
  "_id": "tWYYK4IBDg7ocqbwfqrs",
  "_score": 1,
  "_source": {
    "@timestamp": "2022-07-24T04:55:57Z",
    "news": {
     "author": "News.local",
     "content": null,
     "publishedAt": "2022-07-24T04:55:57Z",
     "description": "今日、素晴らしい出来事がありました！",
     "urlToImage": "https://news.local/hogehoge/hogehoge.jpg",
     "url": "https://news.local/hogehoge.html",
     "title": "これはグッドニュースだ！",
     "source": {
       "id": "News.local",
       "name": "News.local"
     }
    },
    "@version": "1",
    "data_stream": {
     "type": "logs",
     "dataset": "logstash.news",
     "namespace": "qa"
    }
  }
}
```

## まとめ
さて、これまで、Elastic Agent編、Filebeat編、そしてLogstash編と、同じ目的をそれぞれ別の方法で実現する方法を見てきました。色々と方法があって少しややこしいと感じるかもしれませんが、逆にやりたいことに応じて柔軟に実現できる方法があると思って、前向きに捉えたいところです。

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

時系列データのData stream周りのネーミングルールなどが少しややこしいのですが、理解してしまえば、LogstashでもElastic AgentやBeats系と併せて統合的に管理が可能になるので、[公式ドキュメント](https://www.elastic.co/guide/en/fleet/current/data-streams.html#data-streams)を参考にしながら、是非試してみてください。