EC2インスタンスからFluentdを使ってKinesis Data Streamsにログを送信する

## 概要

EC2インスタンスからFluentdを使ってKinesis Data Streamsにログを送信するアーキテクチャを構築します。
![kinesis.png](https://qiita-image-store.s3.amazonaws.com/0/140953/93956efd-eaaf-797b-6447-992cc0b45124.png)
[fluent-plugin-kinesisでKinesis Streamsにログを送信する](https://dev.classmethod.jp/cloud/aws/fluent-plugin-kinesis/)
こちらの記事を参考にしました。
基本的な流れは同じですが、簡単な説明だったり情報が古かったりして少し戸惑ったのでまとめました。
今回は一から作成していますが、必要に応じて作成済みのものを利用してください。

## 環境

作業 OS：macOS Mojave
EC2 OS：Amazon Linux 2 AMI (HVM)
fluentd：1.4.2
td-agent：1.3.3
fluent-plugin-kinesis：3.0.0
ruby：2.6.2

## 目次

1. EC2の作成
2. Streamの作成
3. EC2にロール付与  
    (1) StreamのARNを指定して、Kinesisのポリシーを作成  
    (2) ポリシーを指定して、ロールを作成  
    (3) EC2にロールを付与する  
4. EC2にSSH接続してログイン 
5. Apacheのインストール
6. rbenv/Rubyのインストール 
7. Fluentdのインストール
8. プラグインのインストール 
9. Fluentd設定ファイルを修正する
10. 設定ファイルのシンタックスチェックを行う
11. Fluentdの起動
12. Streamにデータが送信されていることを確認   
    (1) シャードイテレータを確認   
    (2) シャードイテレータを指定してレコード取得  
    (3) デコードして中身を確認  

## 1. EC2の作成
まず、Fluentdを入れるための**AWS EC2**を作成します。
[AWSマネジメントコンソール](https://console.aws.amazon.com/console/home) > EC2 > インスタンス を開きます。
インスタンス作成ボタンから、任意のOSを選択して作成します。
<img width="1143" alt="スクリーンショット 2019-04-03 19.12.53.png" src="https://qiita-image-store.s3.amazonaws.com/0/140953/c92ae6b4-1f2c-5774-0bb7-7544f9eb42f2.png">
今回はAmazon Linuxを利用し、設定等はデフォルトで作成しました。
SHH接続の際に使用するキーペアが生成されるので、保存してください。

## 2. Streamの作成
次に、ログの送信先である**Kinesis Data Stream**を作成します。
[AWSマネジメントコンソール](https://console.aws.amazon.com/console/home) > Kinesis > Data Streamを開きます。
Kibesisストリームの作成ボタンから作成します。
<img width="1143" alt="スクリーンショット 2019-04-03 19.10.16.png" src="https://qiita-image-store.s3.amazonaws.com/0/140953/7a2a434f-1f95-af28-3293-8a8b5db8f561.png">

今回はストリーム名をtest、シャード数を1に設定して作成しました。

※**シャード**…ストリーム内の一意に識別されたデータレコードのシーケンス。ストリームの総容量はシャードの容量の合計。

その後、作成したストリームをクリックし、詳細タブの**ストリーム ARN**の項目を確認しておきます。
これは次のポリシー作成の際に使用します。

<img width="1145" alt="スクリーンショット 2019-04-03 21.28.55.png" src="https://qiita-image-store.s3.amazonaws.com/0/140953/d081c79d-9a36-0ebb-476a-5dd5f503062e.png">

## 3. EC2にロール付与
先ほど作成したストリームのリソースに対するKinesisの権限をEC2に付与します。

#### (1) 作成したStreamのARNを指定して、Kinesisのポリシーを作成

[AWSマネジメントコンソール](https://console.aws.amazon.com/console/home) > IAM > ポリシー を開きます。
ポリシーの作成ボタンから作成します。
<img width="1147" alt="スクリーンショット 2019-04-03 21.39.12.png" src="https://qiita-image-store.s3.amazonaws.com/0/140953/ef020f63-e003-f14e-8368-03a805e19c01.png">
ビジュアルエディタの設定で、以下のように設定します。(その他の設定は任意です)

- サービス：Kinesis
- アクション：すべてのKinesisアクション (kinesis:*)
- リソース：stream > ARNの追加 > ARNを手動でリスト > ストリームARNを貼り付ける

ここではポリシーの名前をtest-kinesis-policyとします。

#### (2) 作成したポリシーを指定して、ロールを作成

[AWSマネジメントコンソール](https://console.aws.amazon.com/console/home) > IAM > ロール を開きます。
ポリシーの作成ボタンから作成します。
<img width="1148" alt="スクリーンショット 2019-04-03 21.42.31.png" src="https://qiita-image-store.s3.amazonaws.com/0/140953/025d7546-0779-fc3a-8637-596426ee90b6.png">
設定は以下のようにします。（その他の設定は任意です。）

- このロールを使用するサービスを選択：EC2
- Attach アクセス権限ポリシー：test-kinesis-policy

ここではロールの名前をtest-kinesis-roleとします。

#### (3) EC2に作成したロールを付与する

[AWSマネジメントコンソール](https://console.aws.amazon.com/console/home) > EC2 > インスタンス を開きます。
作成したEC2を選択 > アクション > インスタンスの設定 > IAMロールの割り当て/置換 を開きます。
<img width="1144" alt="スクリーンショット 2019-04-03 21.47.35.png" src="https://qiita-image-store.s3.amazonaws.com/0/140953/e771e60c-c55c-5798-2f99-21c226e61a14.png">
以下を指定して、適用します。

- IAMロール：test-kinesis-role

## 4.  EC2にSSH接続してログイン 
ターミナルを開いて、EC2にSSH接続します。

```bash
$ ssh -i <key-pair>.pem <ユーザ名>@<パブリックIP or パブリックDNS名>
```

参考：[[初心者向け] 初めてのEC2ログイン：Linux編
](https://dev.classmethod.jp/cloud/aws/first-login-to-ec2-linux/)

## 5. Apacheのインストール
EC2に**Apatch**をインストールして起動します。

```bash
$ sudo yum install -y httpd
$ sudo service httpd start
```

## 6. rbenv/Rubyのインストール 
EC2に**rbevn**と**Ruby**をインストールします。
Rubyのバージョンは今回は安定版の2.6.2(2019/4/3現在)をインストールしました。
※[Fluentdの公式](https://docs.fluentd.org/v1.0/articles/install-by-gem)ではRubyのversionは**2.1以上**をサポートしているとのことです。

参考：[Amazon Linuxに rbenv をインストールして、Rubyをバージョンアップ（Bundlerも）](https://qiita.com/areaz_/items/28692ce46a899291b85b)


## 7. Fluentdのインストール

EC2に**Fluentd**をgemでインストールします。

```bash
$ gem install fluentd --no-document
```

参考：
[Installing Fluentd Using Ruby Gem | Fluentd]( https://docs.fluentd.org/v1.0/articles/install-by-gem)
※Fluentdの公式に書かれている`--no-ri --no-rdoc`オプションは廃止されました。

以下のコマンドで、エラーなく動くことを確認します。

```bash
$ fluentd --setup ./fluent
$ fluentd -c ./fluent/fluent.conf -vv &
$ echo '{"json":"message"}' | fluent-cat debug.test
```

最後のコマンドは、Fluentdに「debug.test」タグのメッセージ「{"json"： "message"}」を送信します。
インストールが成功していた場合、Fluentdは以下のメッセージを出力します。

```bash
2019-04-04 xx:xx:xx +xxxx debug.test：{"json"："message"}
```

## 8. プラグインのインストール
EC2に[fluent-plugin-kinesis]( https://github.com/awslabs/aws-fluent-plugin-kinesis#installation )をインストールします。

```bash
$ sudo td-agent-gem install fluent-plugin-kinesis
```

OSによってはtd-agent-gemコマンドのパスが通っていない場合があるので、通して実行してください。

参考：[Plugin Management | Fluentd](https://docs.fluentd.org/v0.12/articles/plugin-management)

## 9. Fluentd設定ファイルを修正する

Fluentdの設定を行います。

```bash
$ sudo vim /etc/td-agent/td-agent.conf
```

以下をファイルに追記します。
データソースの定義が \<source>タグ、データ処理の定義が \<match>タグです。
使い方などは[aws-fluent-plugin-kinesis#Getting started](https://github.com/awslabs/aws-fluent-plugin-kinesis#getting-started)を確認してください。

```
<source>
  @type tail
  @id output_log
  format apache2
  path /var/log/httpd/access_log
  pos_file /var/log/td-agent/httpd-access.pos
  tag log.httpd.access
</source>
 
<match log.httpd.*>
  @type kinesis_streams
  @id input_httpd
  stream_name test
  region ap-northeast-1
  random_partition_key true
  use_yajl true
</match>
```

## 10. 設定ファイルのシンタックスチェックを行う

```bash
$ td-agent --dry-run -c /etc/td-agent/td-agent.conf 
```

エラーがおきなければOKです。

## 11. Fluentdの起動

Fluentdを起動します。
まずはフォアグラウンドで実行できることを確認して、その後バックグラウンドで実行します。

##### フォアグラウンド実行

```bash
$ sudo td-agent -c /etc/td-agent/td-agent.conf
```

##### バックグラウンド実行

```bash
$ sudo service td-agent start
```

Apacheにリクエストを投げて、Kinesis Streamに送信するログを生成します。

```bash
$ curl localhost
```

## 12. Streamにデータが送信されていることを確認
#### (1) シャードイテレータを確認 

```bash
$ STREAM_NAME=test
$ aws kinesis get-shard-iterator --shard-id shardId-000000000000 --stream-name $STREAM_NAME --shard-iterator-type TRIM_HORIZON
{
    "ShardIterator": "AAAAAAAAAAEYTwTh..."
}
```

#### (2) シャードイテレータを指定してレコード取得

```bash
$ aws kinesis get-records --shard-iterator  "AAAAAAAAAAEYTwTh..."
{
    "Records": [
        {
            "Data": "eyJob3N0IjoiMTI3LjAuMC4xIiwidXNlciI6bnVsbCwibWV0aG9kIjoiR0VUIiwicGF0aCI6Ii8iLCJjb2RlIjo0MDMsInNpemUiOjM2MzAsInJlZmVyZXIiOm51bGwsImFnZW50IjoiY3VybC83LjYxLjEifQo=", 
            "PartitionKey": "8c3e250cc103933ec37f63cec968054c", 
            "ApproximateArrivalTimestamp": 1554283487.951, 
            "SequenceNumber": "49594439765182586642277267931515081312153319763399933954"
        }, 
        {
            "Data": "eyJob3N0IjoiMTI3LjAuMC4xIiwidXNlciI6bnVsbCwibWV0aG9kIjoiR0VUIiwicGF0aCI6Ii8iLCJjb2RlIjo0MDMsInNpemUiOjM2MzAsInJlZmVyZXIiOm51bGwsImFnZW50IjoiY3VybC83LjYxLjEifQo=", 
            "PartitionKey": "10e21500031d1d5f5d6beadb2fbd5fe3", 
            "ApproximateArrivalTimestamp": 1554283679.042, 
            "SequenceNumber": "49594439765182586642277267931516290237972947586714173442"
        }
    ], 
    "NextShardIterator": "AAAAAAAAAAEVvuiUpuBdo5LXOID3x5ksssSyqNfkxhNFkG7jkkhiqaX7knHtVolTj2JgOIgYT0vJVsYgIQA9BT950RdbxFfSIgyv2y1t75br6/UQlZXfVN3TWHLKJPyu9Xzx0fr+TN3avc0UyzOwg++NQCaPEBV2dTaDhXY03Q0dpXPhROAF7Dnkm8oxRjNYD3tIN+mkkPHu1YZs7mAOwVTuew85iAmF", 
    "MillisBehindLatest": 0
}
```

#### (3) デコードして中身を確認

```bash
$ DATA="eyJob3N0IjoiMTI3LjAuMC4xIiwidXNlciI6bnVsbCwibWV0aG9kIjoiR0VUIiwicGF0aCI6Ii8iLCJjb2RlIjo0MDMsInNpemUiOjM2MzAsInJlZmVyZXIiOm51bGwsImFnZW50IjoiY3VybC83LjYxLjEifQo="
$ echo -n $DATA | base64 -d
{"host":"127.0.0.1","user":null,"method":"GET","path":"/","code":403,"size":3630,"referer":null,"agent":"curl/7.61.1"}
```

Apatchから投げたリクエストを確認できました。

## 最後に
記事に間違いや不明点があれば遠慮なくご指摘ください。