Logstashのoutput s3 plugin検証まとめ

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

# 検証構成図
AWSの異なるリージョン間をVPC-Peeringで接続し、NLBを介してログデータをLogstashを使ってS3にアップロードしています。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/211371/13fc6728-fae8-da60-02d2-309cd55b128d.png)
マルチリージョンで構成されたAWS環境の各種サーバのログを1ヶ所のS3バケットに集約してデータレイク化するユースケースに利用出来ると思っています。

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

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

 - [VPCピアリングを作りながら学んでみた][01]
 - [リージョンをまたいでアクセス可能に！NLBがInter-Region VPC Peeringに対応しました][02]
 - [Route53 リゾルバー登場！オンプレミス-VPCの相互名前解決が簡単に実現できるようになりました!][03]

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

# 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ポリシー`は以下の通りです。

```ruby: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. 中間ファイル生成の動作
 2. ファイル圧縮有効化時の動作
 3. ファイルローテーションの動作

## 1. 中間ファイル生成の動作
まずは以下のような`logstash.conf`で`temporary_directory`の動きを見てみます。
`temporary_directory`はデフォルトで`/tmp/logstash`が指定されますので未設定でも動きます。

```ruby: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
```
中間ファイルの命名ルールは以下のようになっています。
>![image.png](https://qiita-image-store.s3.amazonaws.com/0/211371/071ded85-81af-1f2d-60cc-d72e5a3731ef.png)

ファイルローテーション後、`/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最終更新日時とのズレは無視で(笑)
![image.png](https://qiita-image-store.s3.amazonaws.com/0/211371/72c477c4-8c92-dc3c-67f0-8ac490072a8b.png)



## 2. ファイル圧縮有効化時の動作
次は`encoding=>"gzip"`をoutput s3に追記します。

```ruby: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バケット課金の節約になりますね！
![image.png](https://qiita-image-store.s3.amazonaws.com/0/211371/efc7d0c2-87ea-7b2b-5d76-296cd8a2b4b5.png)

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

```ruby: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へ出力されています。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/211371/82b73a12-ee42-6f98-92e1-69ea78f5a20a.png)

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

```ruby: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を超えているというファイルサイズにバラつきが出ていました。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/211371/abbd5c04-aa60-5a6c-4838-5d105b99b5c8.png)

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

# まとめ
基本的な動作は理解することが出来ました。データの流れは以下の通りになります。
![image.png](https://qiita-image-store.s3.amazonaws.com/0/211371/5bdf54f8-b637-5d9d-6cfd-679c09987f50.png)
やっていることは/tmp/logstashにoutput fileしていて、それを定期的に`aws s3 mv`コマンドでS3にアップしている感じの動きでした。
/tmp/logstash配下に生成される中間ファイルを開いてみましたが、inputしたログがそのまま書かれていました。

[01]:https://dev.classmethod.jp/etc/handson-vpc-peering/ "VPCピアリングを作りながら学んでみた"
[02]:https://dev.classmethod.jp/cloud/aws/nlb-access-via-inter-region-vpc-peering/ "リージョンをまたいでアクセス可能に！NLBがInter-Region VPC Peeringに対応しました"
[03]:https://dev.classmethod.jp/cloud/aws/route53-resolver/ "Route53 リゾルバー登場！オンプレミス-VPCの相互名前解決が簡単に実現できるようになりました!"
[04]:https://www.elastic.co/guide/en/logstash/6.5/plugins-outputs-s3.html "S3 output plugin"