初投稿ですが、ただのインストールメモです

今回はAmazon Kinesisに送る

streamを作る

aws kinesis create-stream --stream-name stream --shard-count 2

streamの情報取得

aws kinesis describe-stream --stream-name stream
{
    "StreamDescription": {
        "StreamStatus": "ACTIVE",
        "StreamName": "stream",
        "StreamARN": "arn:aws:kinesis:ap-northeast-1:622601022736:stream/stream",
        "Shards": [
            {
                "ShardId": "shardId-000000000000",
                "HashKeyRange": {
                    "EndingHashKey": "170141183460469231731687303715884105726",
                    "StartingHashKey": "0"
                },
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49562754734384102757616484984742799371518611138497478656"
                }
            },
            {
                "ShardId": "shardId-000000000001",
                "HashKeyRange": {
                    "EndingHashKey": "340282366920938463463374607431768211456",
                    "StartingHashKey": "170141183460469231731687303715884105726"
                },
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49562754734406403502815015607884335089791259500003459096"
                }
            }
        ]
    }
}

fluentdのインストール

curl -L https://toolbelt.treasuredata.com/sh/install-redhat-td-agent2.sh | sh
...
Installed:
  td-agent.x86_64 0:2.3.1-0.el6

Complete!

fluent-plugin-kinesisを使う

td-agent-gem install fluent-plugin-kinesis
Fetching: concurrent-ruby-1.0.2.gem (100%)
Successfully installed concurrent-ruby-1.0.2
Fetching: os-0.9.6.gem (100%)
Successfully installed os-0.9.6
Fetching: middleware-0.1.0.gem (100%)
Successfully installed middleware-0.1.0
Fetching: protobuf-3.6.9.gem (100%)
Successfully installed protobuf-3.6.9
Fetching: fluent-plugin-kinesis-1.0.1.gem (100%)
Successfully installed fluent-plugin-kinesis-1.0.1
Parsing documentation for concurrent-ruby-1.0.2
Installing ri documentation for concurrent-ruby-1.0.2
Parsing documentation for os-0.9.6
Installing ri documentation for os-0.9.6
Parsing documentation for middleware-0.1.0
Installing ri documentation for middleware-0.1.0
Parsing documentation for protobuf-3.6.9
Installing ri documentation for protobuf-3.6.9
Parsing documentation for fluent-plugin-kinesis-1.0.1
Installing ri documentation for fluent-plugin-kinesis-1.0.1
Done installing documentation for concurrent-ruby, os, middleware, protobuf, fluent-plugin-kinesis after 10 seconds
5 gems installed

fluentdの設定

td-agent.conf

<source>
  type tail
  format /^(?<month>[^ ]*) (?<day>[^ ]*) (?<times>[^ ]*) (?<proc>(.+))\[(?<id>[^\]]*)\]: \[(?<level>[^\]]*)\](?<content>(.+))$/
  path /var/log/drive/activedrive-%Y%m%d.log
  pos_file /var/log/td-agent/drive.kinesis.pos
  tag drive.kinesis
</source>

<match drive.kinesis>
  type kinesis
  stream_name stream
  region ap-northeast-1
  random_partition_key true
  use_yajl true
  retry_limit 4
  retry_wait 10s
  flush_interval 1s
  buffer_type file
  buffer_path /var/log/td-agent/kinesis.*.buffer
  buffer_chunk_limit 1024m
  buffer_queue_limit 256
</match>

<filter drive.*>
  type record_transformer
  <record>
    host "#{Socket.gethostname}"
  </record>
</filter>

include_tag_key trueを設定すると出力にtagを追加できる
流れるログはこんなの

Jun 10 15:25:48 adfs[16132:36526832]: [DEV DEBUG]prop_server: quit()`

出力例

{"month":"Jun","day":"10","times":"15:25:48","proc":"adfs","id":"16132:36526832","level":"DEV DEBUG","content":"prop_server: quit()","host":"hostname"}

レコードの取得

とりあえずawscliを使う

aws kinesis get-records --shard-iterator $(aws kinesis get-shard-iterator --stream-name stream --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON|jq .ShardIterator)
{
    "Records": [
        {
            "PartitionKey": "390a8dfb-646b-4013-b730-02c02e77ae66",
            "Data": "eyJtb250aCI6Ikp1biIsImRheSI6IjEwIiwidGltZXMiOiIxNToyNTo0OCIsInByb2MiOiJhZGZzIiwiaWQiOiIxNjEzMjozNjUyNjgzMiIsImxldmVsIjoiREVWIERFQlVHIiwiY29udGVudCI6InByb3Bfc2VydmVyOiBxdWl0KCkiLCJob3N0IjoiaG9zdG5hbWUifQ==",
            "SequenceNumber": "49562754734384102757616484985846548644830444341887500296"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAFgVinmFZpF4IGMtKLEEzxkN+zIUDmjpf+j++pGcb92kMH+RgJ/D9IsV/IhG3dpgEhXZE7V9gdjG5zWABqLL/xr0eWIrtNIM5eGMOWB3Bx5ddrKNKcqFVa99PQD1cUXZehxIa7+2lHh3ZRS3bvXAiKg65SO533JOL/rHLcDCfI0HfVIJ2Wmcg/WwaNU13eaethTF3eHRx6Q6DQmO0aCJh06"
}

awscliではbase64のデコードは提供されていない
KPL,KCLを使用するのが前提になっていると思われる。
KPL
KCL Ruby