kinesis に apache のログを fluentd で流し込む

vpc の設定†

VPC 内で使う場合は endpointを作成する必要がある。
PrivateLink と呼んでいるらしい。

VPC -> エンドポイント -> kinesis を検索
VPC と該当する subnet に attache 
特に セキュリティグループを要注意 443 の inbound を anyware で空けておく必要がある

kinesisの起動†

STREAM=stream1
aws kinesis create-stream --stream-name $STREAM --shard-count 1
aws kinesis wait stream-exists --stream-name $STREAM

fluentd

fluentd install†

curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent3.sh | sh
td-agent-gem install fluent-plugin-kinesis
td-agent-gem install fluent-plugin-stdin --no-document

/etc/td-agent/td-agent.conf

<source>
  @type tail
  format apache2
  path /var/log/httpd/access_log
  pos_file /var/log/td-agent/httpd-access.pos
  tag log.httpd.access
</source>

<source>
  @type stdin
  format csv
  keys tag,time
  tag log.stdin
</source>

<match log.httpd.*>
  @type kinesis_streams
  stream_name stream1
  region ap-northeast-1
  use_yajl true
</match>

チェック

td-agent --dry-run -c /etc/td-agent/td-agent.conf

fluentd をコマンドラインから起動†

 cat hoge.csv | /opt/td-agent/embedded/bin/fluentd -c /etc/td-agent/td-agent.conf

fluentd をサービスとして起動†

sudo td-agent -c /etc/td-agent/td-agent.conf

コマンド†

aws kinesis describe-stream --stream-name 【stream】

データ確認†

wget localhost
STREAM=stream1
watch "aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name $STREAM | jq -r '.ShardIterator' |xargs aws kinesis get-records --shard-iterator | jq '.Records[].Data' |sed 's/\"//g'| base64 --decode |  sed 's/}{/\n/g'"

参考文献†

https://qiita.com/toshihirock/items/86a7d5144ad54b18bf36
https://dev.classmethod.jp/cloud/aws/fluent-plugin-kinesis/