【祝東京上陸】Kinesis Data AnalyticsでCloudWatch Logsをリアルタイム分析する

# はじめに
Amazon Kinesis Data Analytics が[東京リージョンで利用可能](https://aws.amazon.com/jp/blogs/news/amazon-kinesis-data-analytics-is-now-available-in-tokyo-region/)になりました。
[Apache Kafkaのマネージドサービス](
https://aws.amazon.com/jp/blogs/aws/amazon-managed-streaming-for-apache-kafka-msk-now-generally-available/)も発表されたこともあり、ストリーミングサービスの盛り上がりに焦ったので触ってみます。

# Kinesis Data Analyticsとは？
* ストリーミングデータの分析/検索するサービス。
* リアルタイムにフィルターをかけることができるらしい。
* SQLで書けるらしい。

# やってみる
定常的に流れてくるCloudwatch Logsに、Data Analyticsを使って、リアルタイムにフィルターをかけてみます。
今回はCwLogsに出力しているVPCFlowLogsを例にやってみます。

# 流れ
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/7fed0d79-a7c4-360c-161b-e131567bcd61.png)

# 1. Kinesis Stream
* `VPCFlowLogs` という名前で作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/6286eb0b-2381-987d-ab37-161fc94b80ae.png)

## サブスクリプション
CloudWatch LogsをKinesis Streamに連携するには [サブスクリプション](https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/logs/Subscriptions.html) という機能を使います。
マネージメントコンソールでは出来ないようなので、CLIを使います。

### IAMRole作成
* Kinesisストリームにデータを置く権限をCWLogsに付与

```json
% cat << EOF > TrustPolicyForCWL.json
> {
>   "Statement": {
>     "Effect": "Allow",
>     "Principal": { "Service": "logs.ap-northeast-1.amazonaws.com" },
>     "Action": "sts:AssumeRole"
>   }
> }
> EOF

% aws iam create-role --role-name CWLtoKinesisRole --assume-role-policy-document file://./TrustPolicyForCWL.json
{
    "Role": {
        "AssumeRolePolicyDocument": {
            "Statement": {
                "Action": "sts:AssumeRole",
                "Effect": "Allow",
                "Principal": {
                    "Service": "logs.ap-northeast-1.amazonaws.com"
                }
            }
        },
        "RoleId": "AAAAAAAAAAAAAAAAAAA",
        "CreateDate": "2019-06-04T13:35:24Z",
        "RoleName": "CWLtoKinesisRole",
        "Path": "/",
        "Arn": "arn:aws:iam::123456789:role/CWLtoKinesisRole"
    }
}
```
* PutPolicy

```json
% cat PermissionsForCWL.json
{
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "kinesis:PutRecord",
      "Resource": "arn:aws:kinesis:ap-northeast-1:123456789:stream/VPCFlowLogs"
    },
    {
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": "arn:aws:iam::123456789:role/CWLtoKinesisRole"
    }
  ]
}

% aws iam put-role-policy  --role-name CWLtoKinesisRole  --policy-name Permissions-Policy-For-CWL  --policy-document file://~/PermissionsForCWL.json
```
* サブスクリプションフィルタ作成

```sh
% aws logs put-subscription-filter \
    --log-group-name "/aws/vpcflowlogs/" \
    --filter-name "VPCFlowLogsAllFilter" \
    --filter-pattern "[version, account_id, interface_id, srcaddr != "-", dstaddr != "-", srcport != "-", dstport != "-", protocol, packets, bytes, start, end, action, log_status]" \
    --destination-arn "arn:aws:kinesis:ap-northeast-1:123456789:stream/VPCFlowLogs" \
    --role-arn "arn:aws:iam::123456789:role/CWLtoKinesisRole"
```

CwLogsの画面で確認すると、サブスクリプションが表示され、連携されているのがわかりますね。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/cb615bc1-6800-0d4c-9f86-a0d2da3eed0b.png)

kinesis streamに連携されているか、確認してみましょう。

```sh
% aws kinesis get-records --limit 10 --shard-iterator $(aws kinesis get-shard-iterator --stream-name VPCFlowLogs --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON | jq -r ."ShardIterator") | jq -r '.Records[].Data' | base64 -D | zcat
```

# 2. Kinesis Data Analytics
CwLogsとKinesis Streamの連携はできたので、続いてData Analyticsを作成します。

* アプリケーション作成
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/1612cd4e-f9b8-dc53-42cc-6b00d6029ea9.png)

* sourceで作成したkinesis streamを選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/d4247a84-7430-3ec3-4baf-90c16188e452.png)

## pre-processing Lambda function作成
Cwlogsのinputレコードは圧縮されてるので、解凍する必要があります。
pre-processing設定でAWS Lambdaを作成し、解凍してみましょう。

* Lambda functionの `create new`を選択し、新規作成します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/c23e1849-009c-cb8e-87cd-d8287b47ec9a.png)

以前は[blueprintがあった](https://aws.amazon.com/jp/blogs/big-data/preprocessing-data-in-amazon-kinesis-analytics-with-aws-lambda/)ようなのですが、無くなっているので[こちらの記事](https://aws.amazon.com/jp/blogs/big-data/preprocessing-data-in-amazon-kinesis-analytics-with-aws-lambda/)を参考に作成します。

```javascript
'use strict';
console.log('Loading function');
const zlib = require('zlib');

exports.handler = (event, context, callback) => {
    let success = 0; // Number of valid entries found
    let failure = 0; // Number of invalid entries found
    /* Process the list of records */
    const output = event.records.map((record) => {
        /* Data is base64-encoded, so decode here */
        const compressedData = Buffer.from(record.data, 'base64');
        try {
            const decompressedData = zlib.unzipSync(compressedData);
            /* Encode decompressed JSON or CSV */
            const result = (Buffer.from(decompressedData, 'utf8')).toString('base64');
            success++;
            return {
                recordId: record.recordId,
                result: 'Ok',
                data: result,
            };
        } catch (err) {
            failure++;
            return {
                recordId: record.recordId,
                result: 'ProcessingFailed',
                data: record.data,
            };
        }
    });
    console.log('Processing completed.  Successful records ${success}, Failed records ${failure}.');
    callback(null, {
        records: output,
    });
};
```

`Discover schema`実行すると、正しくパースされたデータが表示されましたね。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/381f672b-7dca-d2ed-e805-c0089e2f54c6.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/b0a063c5-4d55-e84b-2658-64d49934edd5.png)


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/5bc58e2c-7711-ffd9-219b-fde52d7a6d05.png)

## SQL作成
Source設定はできたので、実際にフィルターをかけるSQLを作成します。
`Real time analytics` を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/cfab37a6-453b-d5a4-1b63-f5552eead723.png)


byte数でフィルターしてみます。

```sql
-- ** Continuous Filter ** 
-- Performs a continuous filter based on a WHERE condition.
--          .----------.   .----------.   .----------.              
--          |  SOURCE  |   |  INSERT  |   |  DESTIN. |              
-- Source-->|  STREAM  |-->| & SELECT |-->|  STREAM  |-->Destination
--          |          |   |  (PUMP)  |   |          |              
--          '----------'   '----------'   '----------'               
-- STREAM (in-application): a continuously updated entity that you can SELECT from and INSERT into like a TABLE
-- PUMP: an entity used to continuously 'SELECT ... FROM' a source STREAM, and INSERT SQL results into an output STREAM
-- Create output stream, which can be used to send to a destination

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
   "srcaddr"     varchar(16),
   "dstaddr"   varchar(16),
   "bytes"        DOUBLE
);
-- Create pump to insert into output 
CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
SELECT STREAM "srcaddr", "dstaddr", "bytes"
FROM "SOURCE_SQL_STREAM_001"
WHERE "bytes" > 50;
```

お、リアルタイムに流れてきました！(50byteなのであまり確認になってないかもしれませんが、、)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/c08f9b67-5615-78c8-e91f-a19ba7f0ff23.png)

## その後の連携

* `Destination`として、Kinesis stream/Firehose/Lambdaを選択できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/d9bef720-cc9b-7777-04c1-730acf3a219b.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/020aba18-3142-ee1e-6225-1bddfa911e29.png)

Firehoseを使って、フィルター後のデータをelasticsearchでリアルタイムに可視化しする、ってこともできそうですね。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/9038/648d2b4c-7cf6-6121-1c2b-289f7cf68bc8.png)

# まとめ
Kinesis analyticsの挙動を確認してみました。
SQLを使って良い感じにフィルターがかけられそうです！もう少し触ってみようと思います。

(今回の例ではCloudWatch Logs Insightsでも事足りそうですが、、)