はじめに

Logstashのトラブルシューティングでちょいちょい頭を抱えることがあります。
でもなぜかLogstashを使うのを辞めようと思いません(笑)不思議ですね^^

今回はVPCFlowLogsをCloudWatchLogs経由ではなく、より安価なS3経由でElasticsearchに取り込んでみました。

ですが、Filterの書き方でミスってしまい、Elasticsearchに取り込めずにハマってしまい
切り分けのためにlog4j2.propertiesを使ってみたので、備忘録的に記事にしました。

また、公式サイトに記載されていないPluginを初めて使う時のトラブルシューティングでも
有効活用できると思います。

【参考】
・Logstash Logging (Log4j2)

利用環境

product	version
logstash	6.5.4
Elasticsearch	6.5.4
kibana	6.5.4
Java	1.8.0
OS(EC2)	Amazon Linux2 (t3.small)
AMI ID	ami-06b382aba6c5a4f2c
Region	us-east-1

【構成図】

前提として

本投稿では、VPC FlowLogsをElasticsearchに取り込む方法は説明しません。
取り込み方法については、以下の記事を参考にしてみてください。

【参考】
・ VPC FlowLogsをLogstashで正規化してみた

LogstashのLoggingとは

LogstashはJavaで動作していることもあり、Apache Log4j2フレームワークに基づき、ログ出力しています。
Logstashのログは、デフォルト設定で/var/log/logstash配下にINFOレベルで出力されます。

logstash.ymlの抜粋

# ------------ Debugging Settings --------------
#
# Options for log.level:
#   * fatal
#   * error
#   * warn
#   * info (default)
#   * debug
#   * trace
#
# log.level: info
path.logs: /var/log/logstash

/var/log/logstash配下にはlogstash-plain.logとlogstash-slowlog-plain.logの2種類のログが出力されます。

[root@ip-172-31-8-221 logstash]# pwd
/var/log/logstash
[root@ip-172-31-8-221 logstash]# ll
total 21376
-rw-r--r-- 1 logstash logstash 13742951 May  5 22:22 logstash-plain.log
-rw-r--r-- 1 logstash logstash        0 Jan 30 00:43 logstash-slowlog-plain.log

【logstash-plain.log】
・下記のスローログ以外のLogstashに関する全般的なログが記録されます。(←今回のお話)

【logstash-slowlog-plain.log】
・特定のイベントがパイプラインを通過するのに異常に時間がかかる場合にログが記録されます。

実施したこと

以下のinput s3でログの取り込みが出来ていない状況でした。
※後述していますが、このinput s3は悪くなかったです。

logstash.confのinput句の抜粋

input {
  s3 {
    id => "vpcflowlogs"
    bucket => "<S3 Bucket名>"
    prefix => "AWSLogs/<AWSアカウント名>/vpcflowlogs/"
    region => "us-east-1"
    sincedb_path => "/var/lib/logstash/sincedb_vpcflowlogs"
  }
}

そこでLog4j2.properties(/etc/logstash/log4j2.properties)に以下のデバッグ設定を追記してみました。
※input s3のpluginをDEBUGレベルにするという設定
※1行目は、対象のプラグイン名を指定(プラグイン名は、plain.logのログから拾う)
※2行目は、ログレベルを指定

Log4j2.propertiesへの追記内容

# debuglog

logger.s3input.name = logstash.inputs.s3
logger.s3input.level = debug

上記ファイルに設定追加後、Logstashのサービスを再起動します。
再起動後、/var/log/logstash/logstash.plain.logをtail -fしてみました。

[root@ip-172-31-8-221 logstash]# systemctl restart logstash
[root@ip-172-31-8-221 logstash]# tail -f /var/log/logstash/logstash-plain.log 
[2019-05-05T18:54:52,331][WARN ][logstash.runner          ] SIGTERM received. Shutting down.
[2019-05-05T18:54:53,859][INFO ][logstash.pipeline        ] Pipeline has terminated {:pipeline_id=>"vpcflowlogs", :thread=>"#<Thread:0x791ddf9e run>"}
[2019-05-05T18:55:18,189][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.5.4"}
[2019-05-05T18:55:33,982][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@bucket = "<S3 Bucket名>"
[2019-05-05T18:55:33,993][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@id = "vpcflowlogs"
[2019-05-05T18:55:33,993][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@region = "us-east-1"
[2019-05-05T18:55:33,994][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@sincedb_path = "/var/lib/logstash/sincedb_vpcflowlogs"
[2019-05-05T18:55:33,995][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@prefix = "AWSLogs/<AWSアカウント名>/vpcflowlogs/us-east-1/"
[2019-05-05T18:55:33,995][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@enable_metric = true
[2019-05-05T18:55:34,031][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@codec = <LogStash::Codecs::Plain id=>"plain_b7500c07-95c9-4bf3-9adc-bcd9fc4d5bf1", enable_metric=>true, charset=>"UTF-8">
[2019-05-05T18:55:34,033][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@add_field = {}
[2019-05-05T18:55:34,033][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@role_session_name = "logstash"
[2019-05-05T18:55:34,034][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@additional_settings = {}
[2019-05-05T18:55:34,035][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@backup_to_bucket = nil
[2019-05-05T18:55:34,035][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@backup_add_prefix = nil
[2019-05-05T18:55:34,036][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@backup_to_dir = nil
[2019-05-05T18:55:34,037][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@delete = false
[2019-05-05T18:55:34,037][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@interval = 60
[2019-05-05T18:55:34,038][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@watch_for_new_files = true
[2019-05-05T18:55:34,038][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@exclude_pattern = nil
[2019-05-05T18:55:34,039][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@temporary_directory = "/tmp/logstash"
[2019-05-05T18:55:34,039][DEBUG][logstash.inputs.s3       ] config LogStash::Inputs::S3/@include_object_properties = false
[2019-05-05T18:55:34,734][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"vpcflowlogs", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2019-05-05T18:55:37,674][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[<AmazonES FQDN>:443/]}}
[2019-05-05T18:55:37,707][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=><AmazonES FQDN>:443/, :path=>"/"}
[2019-05-05T18:55:37,796][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[<AmazonES FQDN>:443/]}}
[2019-05-05T18:55:37,798][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>https://<AmazonES FQDN>:443/, :path=>"/"}
[2019-05-05T18:55:38,931][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"<AmazonES FQDN>443/"}
[2019-05-05T18:55:38,934][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"<AmazonES FQDN>:443/"}
[2019-05-05T18:55:39,150][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>6}
[2019-05-05T18:55:39,151][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>6}
[2019-05-05T18:55:39,155][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>6}
[2019-05-05T18:55:39,156][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>6}
[2019-05-05T18:55:39,217][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["<AmazonES FQDN>"]}
[2019-05-05T18:55:39,233][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["<AmazonES FQDN>"]}
[2019-05-05T18:55:39,260][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2019-05-05T18:55:39,268][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2019-05-05T18:55:39,931][INFO ][logstash.filters.geoip   ] Using geoip database {:path=>"/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-filter-geoip-5.0.3-java/vendor/GeoLite2-City.mmdb"}
[2019-05-05T18:55:40,333][INFO ][logstash.inputs.s3       ] Registering s3 input {:bucket=>"<S3 Bucket名>", :region=>"us-east-1"}
[2019-05-05T18:55:40,997][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"vpcflowlogs", :thread=>"#<Thread:0x1848e956@/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/create.rb:51 run>"}
[2019-05-05T18:55:41,146][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:vpcflowlogs], :non_running_pipelines=>[]}
[2019-05-05T18:55:41,920][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

Logstashの起動シーケンス中にinput s3の設定内容がログに出力されていました。(config LogStash::Inputs::S3のあたりですね)
S3 Bucketに追加されたオブジェクトを取得しに行っている動作がログに出てきました。

[2019-05-05T18:55:44,858][DEBUG][logstash.inputs.s3       ] S3 input: Adding to objects[] {:key=>"AWSLogs/<AWSアカウント名>/vpcflowlogs/us-east-1/2019/05/05/<AWSアカウント名>_vpcflowlogs_us-east-1_fl-04bce1142756487b3_20190505T0915Z_f9ed343f.log.gz"}
[2019-05-05T18:55:44,860][DEBUG][logstash.inputs.s3       ] objects[] length is:  {:length=>1}
[2019-05-05T18:55:44,920][DEBUG][logstash.inputs.s3       ] S3 input processing {:bucket=>"<S3 Bucket名>", :key=>"AWSLogs/<AWSアカウント名>/vpcflowlogs/us-east-1/2019/05/05/<AWSアカウント名>_vpcflowlogs_us-east-1_fl-04bce1142756487b3_20190505T0915Z_f9ed343f.log.gz"}
[2019-05-05T18:55:44,950][DEBUG][logstash.inputs.s3       ] S3 input: Download remote file {:remote_key=>"AWSLogs/<AWSアカウント名>/vpcflowlogs/us-east-1/2019/05/05/<AWSアカウント名>_vpcflowlogs_us-east-1_fl-04bce1142756487b3_20190505T0915Z_f9ed343f.log.gz", :local_filename=>"/tmp/logstash/<AWSアカウント名>_vpcflowlogs_us-east-1_fl-04bce1142756487b3_20190505T0915Z_f9ed343f.log.gz"}
[2019-05-05T18:55:45,275][DEBUG][logstash.inputs.s3       ] Processing file {:filename=>"/tmp/logstash/<AWSアカウント名>_vpcflowlogs_us-east-1_fl-04bce1142756487b3_20190505T0915Z_f9ed343f.log.gz"}

input s3の動作はデフォルト60秒のIntervalで新規のオブジェクトが追加されていないか確認します。
追加されている場合は、デフォルトで/tmp/logstash配下に圧縮されたファイルをダウンロードします。
ダウンロードしたファイルを解凍してから、Filterに引き渡してパース処理を実行していきます。
まさにその動作がDebugログに出ていました。
Intervalのタイミングでオブジェクトに更新がない場合は、以下のようなログが定期的に出力されていました。

[2019-05-05T18:55:44,850][DEBUG][logstash.inputs.s3       ] S3 input: Found key {:key=>"AWSLogs/<AWSアカウント名>/vpcflowlogs/us-east-1/2019/05/05/<AWSアカウント名>_vpcflowlogs_us-east-1_fl-04bce1142756487b3_20190505T0915Z_b5d7f11f.log.gz"}
[2019-05-05T18:55:44,851][DEBUG][logstash.inputs.s3       ] S3 Input: Object Not Modified {:key=>"AWSLogs/<AWSアカウント名>/vpcflowlogs/us-east-1/2019/05/05/<AWSアカウント名>_vpcflowlogs_us-east-1_fl-04bce1142756487b3_20190505T0915Z_b5d7f11f.log.gz"}

まとめ

うまく処理出来ていなかった原因はFilter句の記載ミス(if分岐で全部Dropしてしまっていましたｗ)でした。
ですが、input句の取り込みが正常に動作していたことでFilter句の修正にスポットを当てられたのが
トラブルシューティング的に良かったので、何かの役に立てばと良いなと思い、投稿してみました^^