2020/01/07 追記：

Python 3.8 / Elasticsearch Service 6.8 対応版にアップデートした記事はこちらです。

CloudFront のアクセスログを Elasticsearch Service 6.8 に取り込むメモ（Lambda Python 3.8 版）

【注意！】（2019/04/17 追記）

私の観測範囲では、2019/04/17 朝より、Lambda 環境上の AWS SDK for Python のバージョンが上がり、Pyhthon 2.7（および 3.6）環境のbotocoreが 1.10 系から 1.12 系に変更されました。

それにより、「BotocoreHTTPSessionがない」というエラーが発生するようになりました。

そこで、一部内容を書き換えてbotocore-1.12を直接使わないようにしています。

この記事は、「ALB/CLBのアクセスログをElasticsearch Service (6.0/6.2) に取り込むメモ」の CloudFront アクセスログ版です。

記事で参照している元記事、クラスメソッドさんの

Ingest Node を使って、より簡単に ELB のアクセスログを AWS Lambda で Elasticsearch に取り込む（Developers.IO）

の終わりに、

Ingest Pipeline の設定を変更するだけで同じ Lambda ファンクションで CloudFront のアクセスログも可視化することができます。

と書かれていますが、少なくとも Elasticsearch Service 6.0（以降）ではうまくいきませんでしたので、そのフォローとして書いています。

※タイトルにある通り、やり方は完全に手抜きです…取り急ぎ用意する必要があったので…。

対応のポイント

ベースは前回の記事通りですが、単純に Ingest Pipeline の設定を変えるだけでは、以下の点で問題が生じます。

ログがタブ区切り（TSV？）なので、そのまま Elasticsearch Service の Ingest Pipeline に送るとエラーが出て怒られる。
ログの先頭にコメント行が2行ほど入るので、そこの取り込みでもエラーが出る（無視しても良いが、気持ち悪い…）。
日付の形式が、Grok がデフォルトで持っているパターン（DATE_US・DATE_EU など）では取り込めない（定義すれば良いのですが、定義した上で更に時刻と結合し、タイムゾーンを…と考えるのが面倒）。

というわけで、今回はほぼ Python コード側で対応してしまいます（本当は Ingest Pipeline 側で何とかするのが筋でしょうが…）。

Ingest Pipeline 設定

CloudFront のログ形式はこちらです。

アクセスログ・ログファイル形式（AWS Documentation » Amazon CloudFront » 開発者ガイド）

また、Grok の設定については、こちらを参考にしました。

CloudFrontのアクセスログをKibanaで可視化する（Developers.IO）

※Logstash設定（Advanced）のログメッセージのパースの部分です。

設定（CloudFront用）

$ curl -H "Content-Type: application/json" -XPUT 'https://XXXXX-XXXXXXXXXXXXXXXXXXXXXXXXXX.ap-northeast-1.es.amazonaws.com/_ingest/pipeline/cflog' -d '{
  "processors": [{
    "grok": {
      "field": "message",
      "patterns":[ "%{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:x_edge_location} (?:%{INT:sc_bytes:int}|-) %{IPORHOST:c_ip} %{WORD:cs_method} %{HOSTNAME:cs_host} %{NOTSPACE:cs_uri_stem} %{INT:sc_status:int} %{GREEDYDATA:referrer} %{GREEDYDATA:User_Agent} %{GREEDYDATA:cs_uri_query} %{GREEDYDATA:cookies} %{WORD:x_edge_result_type} %{NOTSPACE:x_edge_request_id} %{HOSTNAME:x_host_header} %{URIPROTO:cs_protocol} (?:%{INT:cs_bytes:int}|-) %{NUMBER:time_taken:float} %{NOTSPACE:x_forwarded_for} %{NOTSPACE:ssl_protocol} %{NOTSPACE:ssl_cipher} %{WORD:x_edge_response_result_type}" ],
      "ignore_missing": true
    }
  },{
    "remove":{
      "field": "message"
    }
  }, {
    "user_agent": {
      "field": "User_Agent",
      "target_field": "user_agent",
      "ignore_failure": true
    }
  }, {
    "remove": {
      "field": "User_Agent"
    }
  }]
}'

コード変更点（ALB/CLB 版から）

25 ～ 31 行目を書き換えます。

コード変更部分

            if not line.strip().startswith('#'):
                data += '{"index":{"_index":"%s","_type":"log"}}\n' % es_index
                data += '{"message":"%s"}\n' % line.strip().replace('"', '\\"').replace('\t', ' ').replace(' ', 'T', 1).replace(' ', 'Z ', 1)

                if len(data) > 3000000:
                    _bulk(data, awsauth)
                    data = ""

コメント行はスキップする。
タブ区切りを半角スペース区切りに変換する。
日付と時刻をTIMESTAMP_ISO8601で取り込める形式に変換する（タイムゾーンはUTC）。

を行っています。

Lambda 変更点

ログの取り込み元バケットのほか、環境変数の Elasticsearch Service に取り込むときの INDEX プレフィクスと、Ingest Pipeline の名前（下図）を変更します。

以上で、CloudFrontのログも取り込まれるようになるはずです。

※おそらく、Elsaticsearch Service 6.2 でも同じ手順で大丈夫だと思います~~（未検証。近日中に検証予定）。~~→04/08 大丈夫でした。

Elasticsearch Service (6.0/6.2) に取り込んだログ（INDEX）をCuratorで削除するメモ

CloudFrontのアクセスログをElasticsearch Service (6.0/6.2) に取り込むメモ（手抜き編）

対応のポイント

Ingest Pipeline 設定

コード変更点（ALB/CLB 版から）

Lambda 変更点