2020/01/07 追記：
Python 3.8 / Elasticsearch Service 6.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 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）のログメッセージのパースの部分です。
$ 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 大丈夫でした。