kinesis data firehoseでCloudWatchLogsのログを収集する

概要
kinesis firehoseの設定
CloudWatchLogsの設定
参考

概要

AWSのCloudWatchLogsでは、ロググループ単位でログを収集することが出来るが、CloudWatchLogsの分析機能のまま分析するには少々非力であったり、別のBIツールに統合したいものだと思う。

そのため、なんとかしてそのログを取り出すことが必要となるのだが、その方法の一つとしてkinesis data firehoseを使用するものがある。

基本的なやり方は参考を見れば公式でも説明があるので、そちらを見ることでCLIベースの設定について知ることができる。今回はコンソールで作成(一部設定はCLIのみなのでそれは除く)する。

kinesis firehoseの設定

今回の設定では以下の条件を使用する。

ソースはCloudWatchLogs
データの配信先はS3
KMSによる暗号化はしない
最終的な出力形式はJSONLinesにする(JSON末尾に改行が入ったものが複数行記載されているようなフォーマット)

step1: Name and source

[Create Stream]を選択すると、step1から順番に設定が始まる。画面は以下のようになっており…

ここで設定すべきことは、

Stream名
収集するログのソース

の2点である。今回CloudWatchLogsからログを収集するので[Direct Push]を選択する必要がある。

step2: Process Records

このステップではレコードの変換を行う。

入力画面は以下のようになっており…

ここで設定すべきことはデータ変換に関することで、

Lambdaによる変換をするかどうか
Apachのデータベースに取り込めるjson形式にデータを変換するかどうか

上記の2点を設定する。

ここで何を設定すべきか考える。今回ソースにCloudWatchLogsを使用しているが、これを変換なしに使うと次の様なメタデータがついてきてしまう。

{   'logEvents': [   {   'id': '35000659841232929655613*********************************',
                         'message': ' //ここにCloudWatchLogsに蓄積されるログの値',
                         'timestamp': 1569483868348}],
    'logGroup': '*************************************',
    'logStream': '*****************',
    'messageType': 'DATA_MESSAGE',
    'owner': '******************',
    'subscriptionFilters': ['Destination']}

場合によるが、多くの場合CloudWatchLogsに蓄積されるデータは既に所定のJSON形式に変換されており、をそのまま扱いたいのではないのだろうか？今回はそのケースを想定した変換処理(メタデータの削除)を行う。そして変換にはLambda関数を使用するので、[Transform source records with AWS Lambda]をEnabledにする。

そうすると変換に使用するLambda関数を選択するようになるのだが、今回のように単にメタデータを排除するだけならば既に設計図が用意されているのでそれを選択すればよい。

[Create new]を押下すると、以下のように変換用Lambdaに使用できる設計図がいくつか表示される。

このうち赤枠で囲っているものを選択して、Lambda関数を作成する。

Lambda関数の作成では特にコードを変更する必要はないが、以下に記載するようにいくつか設定に注意が必要である。

タイムアウトは5分で設定する。(5分以上はfirehose側でタイムアウトとなるため過度に長くしても意味はない)
メモリはある程度余裕をもたせる。(ケースによるので、これはメトリクスを見ながら調整したい)
IAMロールにはfirehoseのPutRecordBatchとCloudWatchLogsのCreateLogGroup,CreateLogStream,PutLogEventsの権限をもたせる。(CloudWatchLogsはメトリクス監視目的)

Lambda関数作成後、firehoseの設定で選択してあげればここでの設定は終了となる。

step3: Choose a destination

このステップでは変換したデータの保存先について設定する。

保存先については以下の4つから選択することができる。

AWS S3
AWS RedShift
AWS ElasticSearch
Splunk

今回はS3を選択する。

すると画面は以下のようにS3について詳細を設定できる状態となるだろう。

まずS3バケットは特段難しいことはなく、任意のバケットを選択すれば良い。

その下に[S3 prefix]とあるが、これは名前の通り保存されるデータのプレフィックスを制限ないであれば任意のものに設定できる。

サフィックスは指定できず、プレフィックス+<delivery stream name>-<delivery stream version>-<year>-<month>-<day>-<hour>-<minute>-<second>-<uuid><file extension>の固定値によって作成される。例えばデフォルトでは2018/08/10/のように、ディレクトリが分割されるようなプレフィックスが付与されるが、上の画面では既に設定している様に単一のfirehose 名前空間(特殊な変数のようなもの)を設定してやると、 20180810-...のようにディレクトリを分割せずにファイルを生成することも可能である。

docs.aws.amazon.com

一見かなり自由な命名ができそうに見えるが、実際はそれなりに制限があるので詳細は以下を参考にされたい。

docs.aws.amazon.com

[S3 error prefix]については、lambdaによる変換が規定回数を超えて失敗した等なんらかの原因で正しく処理できなかったレコードに付与されるプレフィックスである。詳細は以下を参考にされたい。

最後に[S3 backup]が設定でき、これはLambda関数に渡すレコード郡(バッファしたまとまり)をそのままバックアップする機能である。通常は失敗時のためにEnabledにするところだが、今回は説明のみのためDesableとする。

step4: Configure settings

このステップでは細かい設定をいくつか決めていく。s3の場合以下のような画面が表示されるだろう。

順番に設定していくと…

[S3 buffer conditions]はlambdaではlambdaで変換が終わったレコードを一定の間バッファする制限値を決定する。このときsizeとintervalの2点が設定できるが、これはこの内どちらかの制限値に達した段階でs3への転送が始まるということを意味している。

ちなみに、CloudWatchLogsへのログ出力のスループットが早いと、firehoseは転送に遅れないように動的にこの制限値(主にsize)をスケールして遅れを取り戻そうとする。なのでこの制限値のデータよりもはるかに大きいサイズのデータがS3に出力されていて、システム後段のLambdaのストレージ制限をオーバーしてエラーになった、なんてこともあるのでその点は把握しておきたい。

[S3 compression and encryption]では、S3に置くデータの圧縮と暗号化について設定できる。このうちgzip圧縮については試した事があるのだが、Lambda変換を挟んだ場合は保存されたデータの拡張子に.gzはついているものの実際は圧縮できていないという結果になった。

これはおそらくバグか自分に何らかの問題があるのではないかと考えているので、ひとまず試してダメそうなら外したほうがいいだろう。

ちなみにCloudWatchLogsをソースとして、lambda変換を使わない場合すでにgzip圧縮された状態でS3に置かれる(ただしgz拡張子はついていない…)ので、このオプションでgzipを選択してしまうと2重圧縮となるため注意したい。

[Error logging]については名の通り、S3への転送に関わるエラーが発生した場合、エラーログをCloudWatchLogsに出力する。

また、以下の様にstream作成後のfirehoseのコンソールでもエラー状況は確認できるので状況はかなり見やすかったりする。

Stream作成の最後の設定[Permission]は、IAMroleの選択である。

これについては、初めてfirehoseのstreamを作成する場合は自動で新規作成できるので一旦オートで作成するといいだろう。 [Create new or choose]を押下すると、以下のような画面に遷移する。

ここでプルダウンメニューで[新しいIAMロールの作成]を選択すれば、以下のようなポリシーでIAMロールが作成される。

付与される権限は以下の様になっており、基本的にこれまでの設定とは関係なくkmsやGuluの権限も付与されているため、後から直接ポリシーをいじって権限を適切なものに編集すればいいだろう。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "glue:GetTable",
                "glue:GetTableVersion",
                "glue:GetTableVersions"
            ],
            "Resource": "*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "s3:AbortMultipartUpload",
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::kinesis-cloudwatchlogs-export",
                "arn:aws:s3:::kinesis-cloudwatchlogs-export/*",
                "arn:aws:s3:::%FIREHOSE_BUCKET_NAME%",
                "arn:aws:s3:::%FIREHOSE_BUCKET_NAME%/*"
            ]
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction",
                "lambda:GetFunctionConfiguration"
            ],
            "Resource": "arn:aws:lambda:ap-northeast-1:{{ account }}:function:******************:$LATEST"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:ap-northeast-1:{{ account }}:log-group:/aws/kinesisfirehose/test_firehose_stream:log-stream:*"
            ]
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:GetShardIterator",
                "kinesis:GetRecords"
            ],
            "Resource": "arn:aws:kinesis:ap-northeast-1:{{ account }}:stream/%FIREHOSE_STREAM_NAME%"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kms:Decrypt"
            ],
            "Resource": [
                "arn:aws:kms:ap-northeast-1:{{ account }}:key/%SSE_KEY_ID%"
            ],
            "Condition": {
                "StringEquals": {
                    "kms:ViaService": "kinesis.%REGION_NAME%.amazonaws.com"
                },
                "StringLike": {
                    "kms:EncryptionContext:aws:kinesis:arn": "arn:aws:kinesis:%REGION_NAME%:{{ account }}:stream/%FIREHOSE_STREAM_NAME%"
                }
            }
        }
    ]
}

以上でfirehose側の設定は終了となる。

CloudWatchLogsの設定

最後に、データソースとなるCloudWatchLogsの設定を行う。

LogGroupの作成

これについては特段説明は不要だろうが、収集対象としたいロググループがなければ作成しておく。

サブスクリプションフィルター用のIAMロールを作成する

firehoseへのデータの収集にはサブスクリプションフィルターには適切なIAMロールが必要となる。これは以下の権限を持つポリシーでひとまず十分な挙動が実現できる。

{
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "firehose:*"
            ],
            "Resource": [
                "arn:aws:firehose:ap-northeast-1:{{ account }}:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "iam:PassRole"
            ],
            "Resource": [
                "arn:aws:iam::{{ account }}:role/CWLtoKinesisFirehoseRole"
            ]
        }
    ]
}

この場合IAMのpassroleアクションを付与する必要があるかは微妙(ドキュメントに従い設定している)なので、詳細は以下を参考にされたい。

docs.aws.amazon.com

LogGroupにサブスクリプションフィルターを設定する

これについては唯一コンソールから設定できない作業になるので、aws-cliかaws-sdkのapi経由で設定することとなる。今回はaws-cliから設定することとし、その場合以下のコマンドで設定できる。

aws logs put-subscription-filter \
    --log-group-name //LogGroup名 \
    --filter-name //フィルタ名 \
    --filter-pattern //フィルタパターン(e.g.{$.level = "error"}"(json形式専用のフィルタ構文),""(全てのログを指定)) \
    --destination-arn //kinesis data firehoseのarn \
    --role-arn //先程作成したサブスクリプションフィルター用のIAMrole

なお、CloudWatchLogsのサブスクリプションフィルターは一つのLogGroupに対して一つまでの制限があり、これは緩和できないので注意が必要。