はじめに

Elastic社の提供するETLツールLogstashのoutput s3 pluginについて、動作検証した結果をまとめてみました。

検証構成図

AWSの異なるリージョン間をVPC-Peeringで接続し、NLBを介してログデータをLogstashを使ってS3にアップロードしています。

マルチリージョンで構成されたAWS環境の各種サーバのログを1ヶ所のS3バケットに集約してデータレイク化するユースケースに利用出来ると思っています。

またap-southeast-1を仮想オンプレ環境だと見立てて、オンプレ環境サーバのログもVPNを介して、同様にS3に集約出来るとハイブリッドクラウド環境にも十分活用出来るのではないかと思っています^^

前提

以下のサイトを参考にVPC環境を構築しています。

Logstash S3 Outputとは

Logstashの取り込んだログデータの出力先としてAWS S3バケットを指定することが出来ます。
S3出力を実現するのがこのoutput s3 pluginsになります。

S3 Output Configuration Options

S3 Outputにおける設定項目をまとめてみました。

設定項目	説明	デフォルト値
access_key_id	S3に書き込み権限のあるIAMアカウントで作成したアクセスキーを指定します。	- (任意)
additional_settings	S3に接続する時に利用できるクライアントAPIパラメータを指定します。	{} (任意)
aws_credentials_file	明示的にキー指定しないでAWS認証情報を参照する場合のファイルパスを指定します。	- (任意)
bucket	出力先のS3バケット名を指定します。	- (必須)
canned_acl	S3リソースへのアクセスリストを指定します。	- (任意)
encoding	出力するファイルのgzip圧縮設定をします。	- (任意)
endpoint	VPCエンドポイント経由のアクセス時のリージョン名を指定します。	- (任意)
prefix	指定したS3バケット配下のパスを指定します。	"" (任意)
proxy_uri	Webプロキシ経由でS3に出力する場合はプロキシのURIを指定します。	- (任意)
region	出力S3バケットのAWSリージョンを指定します。	"us-east-1" (任意)
restore	Logstash異常終了時の回復機能利用有無を指定します。	true (任意)
role_arn	AssumeRoleを利用する場合のARNを指定します。	- (任意)
role_session_name	AssumeRoleを利用する場合のセッション名を指定します。	"logstash" (任意)
rotation_strategy	出力するファイルをローテーションするアルゴリズムを指定します。	"size_and_time" (任意)
secret_access_key	アクセスキーを利用する場合の対になるシークレットキーを指定します。	- (任意)
server_side_encryption	サーバ側(S3)でのファイル暗号化設定に合わせて指定します。	false (任意)
server_side_encryption_algorithm	ファイル暗号化している場合の暗号化アルゴリズムを指定します。	"AES256" (任意)
session_token	AWSセッショントークン(一時的な認証情報)を指定します。	- (任意)
signature_version	AWS APIリクエスト署名を利用する場合の署名バージョンを指定します。	- (任意)
size_file	ローテーションするファイルサイズ(byte)を指定します。	5242880 (任意)
ssekms_key_id	サーバ側でAWS KMSを利用した暗号化を実施している場合、KMSキーを指定します。	- (任意)
storage_class	ファイル出力先のS3ストレージクラスを指定します。	"STANDARD" (任意)
temporary_directory	ローテーションされるまでの中間ファイルを配置するパスを指定します。	"/tmp/logstash" (任意)
time_file	ローテーションするタイミング(分)を指定します。	15 (任意)
upload_queue_size	中間ファイルをS3バケットにアップする時に利用するキューサイズを指定します。	4 (任意)
upload_workers_count	中間ファイルをS3バケットにアップする時のワーカー数を指定します。	4 (任意)
validate_credentials_on_root_bucket	S3のルートバケットに対する認証情報のチェック有無を指定します。	true (任意)

Logstashに適用するIAMポリシー

LogstashがS3バケットにログを出力するためにはS3バケットに対する書き込み権限が必要になります。
LogstashがインストールされているEC2に適用したIAMロールに付与したIAMポリシーは以下の通りです。

IAMポリシー

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "s3:PutAnalyticsConfiguration",
                "s3:PutAccelerateConfiguration",
                "s3:DeleteObjectVersion",
                "s3:ReplicateTags",
                "s3:RestoreObject",
                "s3:CreateBucket",
                "s3:ReplicateObject",
                "s3:PutEncryptionConfiguration",
                "s3:DeleteBucketWebsite",
                "s3:AbortMultipartUpload",
                "s3:PutBucketTagging",
                "s3:PutLifecycleConfiguration",
                "s3:PutObjectTagging",
                "s3:DeleteObject",
                "s3:DeleteBucket",
                "s3:PutBucketVersioning",
                "s3:DeleteObjectTagging",
                "s3:PutMetricsConfiguration",
                "s3:PutReplicationConfiguration",
                "s3:PutObjectVersionTagging",
                "s3:DeleteObjectVersionTagging",
                "s3:PutBucketCORS",
                "s3:PutInventoryConfiguration",
                "s3:PutObject",
                "s3:PutBucketNotification",
                "s3:PutBucketWebsite",
                "s3:PutBucketRequestPayment",
                "s3:PutBucketLogging",
                "s3:ReplicateDelete"
            ],
            "Resource": [
                "arn:aws:s3:::<Bucket名>",
                "arn:aws:s3:::<Bucket名>/<Prefix名>/*"
            ]
        }
    ]
}

権限が足りないとlogstash起動時に/var/log/logstash/logstash-plain.logに以下のエラーメッセージが出ますので、適切な権限設定をしましょう。検証であればAmazonS3FullAccessを利用しても良いと思います。

Jan 19 11:54:44 ip-172-31-4-44 logstash: [2019-01-19T11:54:44,284][ERROR][logstash.outputs.s3      ] Error validating bucket write permissions! {:message=>"Access Denied", :class=>"Aws::S3::Errors::AccessDenied", :backtrace=>["/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/seahorse/client/plugins/raise_response_errors.rb:15:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/aws-sdk-core/plugins/s3_sse_cpk.rb:19:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/aws-sdk-core/plugins/s3_dualstack.rb:24:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/aws-sdk-core/plugins/s3_accelerate.rb:34:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/aws-sdk-core/plugins/jsonvalue_converter.rb:20:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/aws-sdk-core/plugins/idempotency_token.rb:18:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/aws-sdk-core/plugins/param_converter.rb:20:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/seahorse/client/plugins/response_target.rb:21:in `call'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/seahorse/client/request.rb:70:in `send_request'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-core-2.11.177/lib/seahorse/client/base.rb:207:in `block in define_operation_methods'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-resources-2.11.177/lib/aws-sdk-resources/services/s3/file_uploader.rb:42:in `block in put_object'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-resources-2.11.177/lib/aws-sdk-resources/services/s3/file_uploader.rb:52:in `open_file'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-resources-2.11.177/lib/aws-sdk-resources/services/s3/file_uploader.rb:41:in `put_object'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-resources-2.11.177/lib/aws-sdk-resources/services/s3/file_uploader.rb:34:in `upload'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/aws-sdk-resources-2.11.177/lib/aws-sdk-resources/services/s3/object.rb:252:in `upload_file'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-output-s3-4.1.7/lib/logstash/outputs/s3/write_bucket_permission_validator.rb:43:in `upload_test_file'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-output-s3-4.1.7/lib/logstash/outputs/s3/write_bucket_permission_validator.rb:18:in `valid?'", "/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-output-s3-4.1.7/lib/logstash/outputs/s3.rb:202:in `register'", "org/logstash/config/ir/compiler/OutputStrategyExt.java:102:in `register'", "org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:46:in `register'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:242:in `register_plugin'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:253:in `block in register_plugins'", "org/jruby/RubyArray.java:1734:in `each'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:253:in `register_plugins'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:594:in `maybe_setup_out_plugins'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:263:in `start_workers'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:200:in `run'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:160:in `block in start'"]}
Jan 19 11:54:44 ip-172-31-4-44 logstash: [2019-01-19T11:54:44,310][ERROR][logstash.pipeline        ] Error registering plugin {:pipeline_id=>"main", :plugin=>"#<LogStash::OutputDelegator:0x6a49a4e1>", :error=>"Logstash must have the privileges to write to root bucket `<Bucket名>`, check your credentials or your permissions.", :thread=>"#<Thread:0x65b401a7 run>"}
Jan 19 11:54:44 ip-172-31-4-44 logstash: [2019-01-19T11:54:44,320][ERROR][logstash.pipeline        ] Pipeline aborted due to error {:pipeline_id=>"main", :exception=>#<LogStash::ConfigurationError: Logstash must have the privileges to write to root bucket `<Bucket名>`, check your credentials or your permissions.>, :backtrace=>["/usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-output-s3-4.1.7/lib/logstash/outputs/s3.rb:203:in `register'", "org/logstash/config/ir/compiler/OutputStrategyExt.java:102:in `register'", "org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:46:in `register'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:242:in `register_plugin'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:253:in `block in register_plugins'", "org/jruby/RubyArray.java:1734:in `each'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:253:in `register_plugins'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:594:in `maybe_setup_out_plugins'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:263:in `start_workers'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:200:in `run'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:160:in `block in start'"], :thread=>"#<Thread:0x65b401a7 run>"}
Jan 19 11:54:44 ip-172-31-4-44 logstash: [2019-01-19T11:54:44,349][ERROR][logstash.agent           ] Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Create<main>, action_result: false", :backtrace=>nil}

検証内容

今回は以下の基本的な動作を検証しました。

中間ファイル生成の動作
ファイル圧縮有効化時の動作
ファイルローテーションの動作

1. 中間ファイル生成の動作

まずは以下のようなlogstash.confでtemporary_directoryの動きを見てみます。
temporary_directoryはデフォルトで/tmp/logstashが指定されますので未設定でも動きます。

logstash.conf

input {
  beats {
    port => 5044
  }
}

output {
  s3 {
    region => "ap-northeast-1"
    bucket => "<Bucket名>"
    prefix => "<Prefix名>/%{+YYYY}/%{+MM}/%{+dd}"
    temporary_directory => "/tmp/logstash_s3"
  }
}

logstash起動後、ファイルローテーションされる前にtemporary_directoryを確認してみました。
1f37b020-6976-4306-9ee2-69ac54f84a51というディレクトリが生成され、その配下にoutput s3のprefixで指定したディレクトリパスが生成されていました。

[root@ip-172-31-4-44 19]# pwd
/tmp/logstash_s3/1f37b020-6976-4306-9ee2-69ac54f84a51/<Prefix名>/2019/01/19
[root@ip-172-31-4-44 19]# ll
total 80
-rw-r--r-- 1 logstash logstash 78882 Jan 19 14:28 ls.s3.08f863e0-4f55-4e29-9a07-2eae56171928.2019-01-19T14.28.part0.txt

中間ファイルの命名ルールは以下のようになっています。

ファイルローテーション後、/tmp/logstash_s3/1f37b020-6976-4306-9ee2-69ac54f84a51/<Prefix名>/2019/01/19配下でllコマンドを実施してみたところ、ファイルがtotal 0という結果が返ってきました。おかしいと思い、temporary_directoryである/tmp/logstash_s3/まで移動し、ディレクトリ状態を確認したところ、152dcfdb-f4fe-493c-94c7-91b85ec99824というディレクトリ名に変わっているではないか...

[root@ip-172-31-4-44 19]# ll
total 0
[root@ip-172-31-4-44 19]# pwd
/tmp/logstash_s3/1f37b020-6976-4306-9ee2-69ac54f84a51/<Prefix名>/2019/01/19
[root@ip-172-31-4-44 19]# cd /tmp/logstash_s3/
[root@ip-172-31-4-44 logstash_s3]# ll
total 0
drwxr-xr-x 3 logstash logstash 22 Jan 19 14:43 152dcfdb-f4fe-493c-94c7-91b85ec99824　　←変わっていたディレクトリ名

ファイルローテーションは、デフォルト値(ファイルサイズが5MBに達するか、15分経過するか、いずれか先に到達したタイミング)としていましたので、15分でローテーションされていました。

ファイルローテーションの都度、temporary_directory直下のディレクトリ名は変わるようです。
・1個目: 1f37b020-6976-4306-9ee2-69ac54f84a51
・2個目: 152dcfdb-f4fe-493c-94c7-91b85ec99824

ファイルはちゃんと指定したS3バケット配下のPrefixに配置されていますね。
※OSのTimezoneをUTCとしていたので、S3最終更新日時とのズレは無視で(笑)

2. ファイル圧縮有効化時の動作

次はencoding=>"gzip"をoutput s3に追記します。

logstash.confのoutput区

output {
  s3{
    region => "ap-northeast-1"
    bucket => "<Bucket名>"
    prefix => "<Prefix名>/%{+YYYY}/%{+MM}/%{+dd}"
    temporary_directory => "/tmp/logstash_s3"
    #以下圧縮設定を追記
    encoding => "gzip"
  }
}

上記confでlogstashを起動します。
/tmp/logstash_s3/87fadc2b-71c2-4000-b33a-794ded6cd61f/<Prefix名>/2019/01/19配下に.gz形式のファイルが生成されていますね。

[root@ip-172-31-4-44 19]# cd /tmp/logstash_s3/
[root@ip-172-31-4-44 logstash_s3]# ll
total 0
drwxr-xr-x 3 logstash logstash 22 Jan 19 15:48 87fadc2b-71c2-4000-b33a-794ded6cd61f
[root@ip-172-31-4-44 logstash_s3]# cd 87fadc2b-71c2-4000-b33a-794ded6cd61f/<Prefix名>/2019/01/19
[root@ip-172-31-4-44 19]# ll
total 4
-rw-r--r-- 1 logstash logstash 940 Jan 19 15:50 ls.s3.0d889839-2648-45b7-a3cf-12d9f0ef2338.2019-01-19T15.48.part0.txt.gz　←中間ファイルが圧縮されている

S3バケット課金の節約になりますね！

3. ファイルローテーションの動作

最後にsize_fileとtime_fileの動きを確認してみました。
以下、デフォルト値から5分間隔でファイルローテーションするようにconfを設定変更します。

logstash.confのoutput区

output {
  s3{
    region => "ap-northeast-1"
    bucket => "<Bucket名>"
    prefix => "<Prefix名>/%{+YYYY}/%{+MM}/%{+dd}"
    temporary_directory => "/tmp/logstash_s3"
    encoding => "gzip"
    #以下time_file設定(15分→5分)を追記
    rotation_strategy => "time"
    time_file => 5
  }
}

logstash起動後、1:16 → 1:21 → 1:26と5分間隔でS3へ出力されています。

最後にファイルサイズ200Bでファイルローテーションするようにconfを設定変更します。

logstash.confのoutput区

output {
  s3{
    region => "ap-northeast-1"
    bucket => "<Bucket名>"
    prefix => "<Prefix名>/%{+YYYY}/%{+MM}/%{+dd}"
    temporary_directory => "/tmp/logstash_s3"
    encoding => "gzip"
    #以下設定変更(size_fileで5MB→200B)
    rotation_strategy => "size"
#    time_file => 5
    size_file => 200
  }
}

ファイルサイズが一定とならず200Bを超えているというファイルサイズにバラつきが出ていました。

ちなみにsystemctl restart logstashコマンドで再起動すると中間ファイルはsize_fileとtime_fileの値に関係なく、停止時点のファイル状態でS3にアップされました。
またsystemctl stop logstashで停止してからsystemctl start logstashで起動するとファイル名のpart部分は0に戻りますが、systemctl restart logstashだと0に戻りませんでした。

まとめ

基本的な動作は理解することが出来ました。データの流れは以下の通りになります。

やっていることは/tmp/logstashにoutput fileしていて、それを定期的にaws s3 mvコマンドでS3にアップしている感じの動きでした。
/tmp/logstash配下に生成される中間ファイルを開いてみましたが、inputしたログがそのまま書かれていました。