EFKでDockerのログを集める usec/nsec対応版

# 概要

- Dockerのログを集約するためElasticSearch + Fluentd + Kibanaを構築した
- しかしESがmsecまでしか対応しておらず、Kibanaで表示した時にログの順番が狂うことがある
    - (msec単位で同じ時刻にログが出た場合)
- Kibanaで正しい表示順になるように環境を構築する

# ポイント

- fluentdは1.0を使う
    - 現在stableの0.12はそもそもsubsecondに対応していない。1.0でnanosecまで対応している。
- dockerからfluentdへの転送はdockerのfluentd logging driverを使う。tailでも可能だが、コンテナ名でのタグ付け等はlogging driverを使ったほうが簡単にできるっぽいため。
- dockerのfluentd logging driver[だけが未だsubsecond対応していない。](https://github.com/moby/moby/issues/17181#issuecomment-332271872)ので、同一ホスト上にまずdockerdから受け取るfluentdをコンテナで起動しておき(fluentd-agent)、この中でrecord_transformerプラグインを使って`@timestamp`を **受信時刻で** 上書きし、これをEFKスタックに送る。この時点で時刻はnanosec精度になる。
    - こうしておけば、将来的にfluentd logging driverがsubsecond対応した際に、レコードのフォーマットは変えずに済むはず。
    - またfluentd logging driverを使うと`docker logs`コマンドが使えなくなる問題にもある程度は対応できる。fluentd-agentのstdoutにdockerdから受けたログを出すようにしておけば、見づらいが一応見ることは可能。
- しかしこれだけではESに入れた時にmillisecまで丸められてしまうため、別途ソート用に `sec.nanosec` 形式のフィールドを作っておく。
    - Kibanaでは2カラムのソートに対応していない。よって1カラムにusec精度でソート可能なデータを入れておく必要あり。
    - record_transformerで生成したnanosec精度の時刻は、dockerd上でログが生成されたオリジナルの時刻から若干遅れが生じるが、同一ホスト上のfluentdで受信した時刻になるので、そこまでズレは無いと考える。それよりもKibana上で見た時に順番が狂ってるほうが運用的にはツラい。
- EFKスタックのfluentdで受けたログは、fluentd-elasticsearch pluginでESに突っ込む。このときフィールドの型はESによって動的に決められるが、ソート用のフィールド `sec.nanosec` はscaled_float形式にしたいため、fluentd-elasticsearch pluginのオプションでESのテンプレートファイルを渡してあげる。

## 現状での各コンポーネントのsubsecond対応状況まとめ

- fluentd → OK
    - 1.0でnanosecまで対応
- elasticsearch → OK?
    - [現状millisecまで](https://github.com/elastic/elasticsearch/issues/10005)
- docker fluentd logging driver → **NG**
    - [subsecond未対応](https://github.com/moby/moby/issues/17181)


# セットアップ

とりあえす `docker-compose.yml` から。デモ用のセットアップで、以下の5つのコンテナが立ち上がる。

- httpd
    - デモ用のWebサーバー。logging driverとしてfluentdを指定。
- fluentd-agent
    - fluentd agent。ログを送りたいコンテナと同一ホスト上で動かす。一つの`docker-compose.yml`毎にfluentd agentを起動するデザインに対応するため、ポート番号をあえてデフォルトの24224からずらしている。
- elasticsearch
- fluentd
- kibana
    - EFKスタック。fluentd-agentから転送されたログをfluentdで受けて、elasticsearchに突っ込んでいる。

実際には、動かすサービス+fluend-agent、EFKスタックの2つに分けて起動して使うことになる。

```yaml:docker-compose.yml
version: '3'

services:

  web:
    image: httpd
    ports:
      - 8080:80
    networks:
      - front-tier
    logging:
      driver: fluentd
      options:
        fluentd-address: localhost:24225
    depends_on:
      - fluentd-agent

  fluentd-agent:
    image: fluent/fluentd:v1.0
    volumes:
      - ./fluentd-agent/etc:/fluentd/etc
    ports:
      - 24225:24225
      - 24225:24225/udp
    networks:
      - logging-tier

  fluentd:
    build: ./images/fluentd
    volumes:
      - ./fluentd/etc:/fluentd/etc
    networks:
      - logging-tier
      - efk-tier
    depends_on:
      - elasticsearch

  elasticsearch:
    image: elasticsearch
    ports:
      - 9200:9200
    networks:
      - efk-tier
    volumes:
      - elasticsearch:/usr/share/elasticsearch/data

  kibana:
    image: kibana
    ports:
      - 5601:5601
    networks:
      - efk-tier

networks:
  front-tier:
  logging-tier:
  efk-tier:

volumes:
  elasticsearch:
```

`fluentd` サービスでbuildしているのはelasticsearch pluginをインストールするため。Dockerfileの中身は極めてシンプル。

```file:./images/fluentd/Dockerfile
FROM fluent/fluentd:v1.0
RUN gem install fluent-plugin-elasticsearch --no-rdoc --no-ri
```

続いてfluentd-agentのfluent.conf

```file:./fluentd-agent/etc/fluent.conf
<source>
  @type forward
  port 24225 # (1)
  bind 0.0.0.0
</source>
<filter *.**>
  @type record_transformer
  enable_ruby
  <record>
    ts_original ${ time.utc.iso8601(9) } # (2)
    ts ${ Time.now.utc.strftime('%s.%N') } # (3)
  </record>
</filter>
<filter *.**>
  @type record_transformer
  enable_ruby
  <record>
    @timestamp ${ Time.strptime(record['ts'], '%s.%N').iso8601(9) } # (4)
  </record>
</filter>
<match *.**>
  @type copy
  <store>
    @type stdout # (5)
  </store>
  <store>
    @type forward
    <server>
      host fluentd
      port 24224
    </server>
    <buffer>
      flush_mode immediate
    </buffer>
  </store>
</match>
```

- (1) ポート番号を変えている
- (2) オリジナルの時刻を`ts_original`フィールドにコピーしておく
- (3) Time.now = 受信した時刻を`ts`フィールドに入れておく。この時形式を `unixtime(sec).nanosec` の文字列にしておく。後でKibanaで表示する時に、このフィールドをソートに使う。
- (4) `@timestamp`を(2)で入れた時刻から復元してiso8601形式で入れる。引数の`9`は小数点以下9桁、つまりnanosecまで含めるという意味。こうすることで(2)と(3)の時刻が全く同一になる。
- (5) `docker logs`の代替で使えるようstdoutにも出しておく

続いてEFKスタック側のfluentdのfluent.conf

```file:./fluentd/etc/fluent.conf
<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>
<match *.**>
  @type copy
  <store>
    @type elasticsearch
    host elasticsearch
    port 9200
    logstash_format true
    include_tag_key true
    tag_key @log_name
    template_name fluentd
    template_file /fluentd/etc/es-template.json
    <buffer>
      flush_mode immediate
    </buffer>
  </store>
  <store>
    @type stdout
  </store>
</match>
```

と、ElasticSearchのテンプレートファイル。

```file:./fluentd/etc/es-template.json
{
  "template": "logstash-*",
  "version": 50001,
  "settings": {
    "index.refresh_interval": "5s"
  },
  "mappings": {
    "_default_": {
      "_all": {
        "enabled": true,
        "norms": false
      },
      "dynamic_templates": [
        {
          "message_field": {
            "path_match": "message",
            "match_mapping_type": "string",
            "mapping": {
              "type": "text",
              "norms": false
            }
          }
        },
        {
          "string_fields": {
            "match": "*",
            "match_mapping_type": "string",
            "mapping": {
              "type": "text",
              "norms": false,
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            }
          }
        }
      ],
      "properties": {
        "@timestamp": {
          "type": "date",
          "include_in_all": false
        },
        "@version": {
          "type": "keyword",
          "include_in_all": false
        },
        "geoip": {
          "dynamic": true,
          "properties": {
            "ip": {
              "type": "ip"
            },
            "location": {
              "type": "geo_point"
            },
            "latitude": {
              "type": "half_float"
            },
            "longitude": {
              "type": "half_float"
            }
          }
        },
        "ts": { // (6)
          "type": "scaled_float",
          "scaling_factor": 1000000000
        }
      }
    }
  }
}
```

中身は[logstashのElasticSearchプラグインのテンプレート](https://github.com/logstash-plugins/logstash-output-elasticsearch/blob/master/lib/logstash/outputs/elasticsearch/elasticsearch-template-es5x.json)に、(5)の定義だけを追加したもの。

- (6) `ts`プロパティに`scaled_float`タイプを指定。`ts`には値として`sec.usec`という文字列で値が入っているので、これを数値にパースしてあげる。[scaling_factorは小数点以下の桁数を指定するもの](https://www.elastic.co/guide/en/elasticsearch/reference/current/number.html#scaled-float-params)で、ここでは9桁あるので1,000,000,000を指定している。これで、このプロパティでソートが可能になり、usec精度でのソートが可能となる。

http://localhost:5601 でKibanaのコンソールにアクセスし、index patternを設定。Time Filter filed nameは`@timestamp`にしておく。デモ用のWebサーバー http://localhost:8080 に何回かアクセスしてからDiscoveryタブに行くと、ログが表示されている。表示カラムにtsを追加し、ソートができることを確認。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/28237/f3e3b31d-ea5e-0efb-f63f-acce197e7c66.png)

# Tips

- `ts` カラムの表示がちょっと幅を取りすぎているので、調整する。Management->Index Patternsで`ts`をedit(右端のボタン)、FormatでNumberを選択しpatternに`.00000`と入れてあげると、小数点以下5桁だけを表示するようになる。桁数はお好みで。ちなみにStringを選んであげると、小数点9桁、つまりnsecまでちゃんとデータが入っていることが確認できる。