CloudWatch LogsのログをRedshiftやAthenaで分析する

# 目的
CloudWatch Logsのログを分析したい。

Kinesis Firehoseを使えばS3に転送できるが、そのままだと{json}{json}のように１行に複数のJSONオブジェクトが保存されてしまう。
どうやら自力で
Base64デコード=>gunzip=>logEventsを取り出す=>中のmessageに改行文字を追加して結合=>Base64エンコードして返り値として渡す
という処理が必要らしい

https://aws.typepad.com/sajp/2017/03/aws-black-belt-online-seminaramazon-athena_slides_and_qa.html

というわけで順番にやっていきます

## CloudWatch Logsの作成
### Log Groupの作成
CloudWatchのLogsから
Actions -> Create log group
![スクリーンショット 2019-05-09 18.13.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/36192/fc8fc9da-bb63-485d-836c-0a323a50b408.png)

### Log Streamの作成
上で作ったLog GroupをクリックしてCreate Log Stram
![スクリーンショット 2019-05-09 18.15.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/36192/4334611c-b9a0-870c-07d3-843b30bc03d8.png)

### 試しに流してみる

``` put-log-events.sh
#!/bin/bash
 
export AWS_DEFAULT_REGION=ap-northeast-1
 
LogGroupName="redshift-experiment"
LogStreamName="command-stream"
 
Profile=$1
Mess=$2
 
Mess=`echo $Mess | sed -e "s/'//g"`
 
UploadSequenceToken=$(aws --profile $Profile logs describe-log-streams --log-group-name "$LogGroupName" --query 'logStreams[?logStreamName==`'$LogStreamName'`].[uploadSequenceToken]' --output text)
 
TimeStamp=`gdate "+%s%N" --utc`
TimeStamp=`expr $TimeStamp / 1000000`
 
if [ "$UploadSequenceToken" != "None" ]
then
  aws --profile $Profile logs put-log-events --log-group-name "$LogGroupName" --log-stream-name "$LogStreamName" --log-events timestamp=$TimeStamp,message="$Mess" --sequence-token $UploadSequenceToken
else
  aws --profile $Profile logs put-log-events --log-group-name "$LogGroupName" --log-stream-name "$LogStreamName" --log-events timestamp=$TimeStamp,message="$Mess"
fi
```

事前にaws cliをインストールして設定しておいてください。
あと```$ brew install coreutils```でgdateを使えるようにしておきましょう。
awsアカウントって大抵複数持ってるだろうから、、[こちら](https://dev.classmethod.jp/cloud/aws/put-cloudwatchlogs/)のスクリプトをコピーしてprofile渡せるようにしたり、macで動くようにdateをgdateに変えた感じです。

```$ ./put-log-events.sh default hello```

でログが増えていきます。

## Kinesis Streamの作成
cloudwatch-logs-streamというstremを作ります。
お金は大切なので出来るだけ小さく作りましょう。
![スクリーンショット 2019-05-09 18.24.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/36192/f8957bab-749a-bf5c-bd62-ad8351fb2b04.png)

### roleの作成
[こちら](https://qiita.com/tilfin/items/150cd899077d3c6b7edd)がとても参考になります。

```cwl-role.json
{
  "Statement": {
    "Effect": "Allow",
    "Principal": { "Service": "logs.ap-northeast-1.amazonaws.com" },
    "Action": "sts:AssumeRole"
  }
}
```

```cwl-kinesis-write.json
{
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "kinesis:PutRecord",
      "Resource": "arn:aws:kinesis:ap-northeast-1:xxxx:stream/cloudwatch-logs-stream"
    },
    {
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": "arn:aws:iam::xxxx:role/cwl-to-kinesis"
    }
  ]
}

```

```
$ aws iam create-role --role-name cwl-to-kinesis --assume-role-policy-document file://./cwl-role.json --profile default
```

```
$ aws iam put-role-policy --role-name cwl-to-kinesis --policy-name cwl-kinesis-write --policy-document file://./cwl-kinesis-write.json --profile default
```

### サブスクリプションフィルタを作成

```
aws logs \
    put-subscription-filter \
    --region ap-northeast-1 \
    --log-group-name "redshift-experiment" \
    --filter-name "All" \
    --filter-pattern "" \
    --role-arn "arn:aws:iam::xxxx:role/cwl-to-kinesis" \
    --destination-arn "arn:aws:kinesis:ap-northeast-1:xxxx:stream/cloudwatch-logs-stream" \
    --profile default
```

## S3へ保存
S3に保存するためにKinesis Firehoseの設定
cloudwatch-logs-delivery-streamを作ります

Amazon KinesisのData FirehoseでCreate Delivery Streamをクリック
Step1でSourceにさっき作ったKinesis Streamを選択


![スクリーンショット 2019-05-09 18.45.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/36192/78c8eb23-ad77-2cde-ec62-1a5aacf378b1.png)

Step2　デフィオルトでOK

Step3 DestinationでS3を選択保存先のbucketやprefixは適当につけておきましょう。

Step4は基本デフォルトでOK。ただしStep4のBuffer intervalだけ60Secondを指定しておなかないとログを流してもなかなか反映されません。
IAM roleは検証環境とかであれば新しく作ってしまいましょう。

これで作ってログを流すとs3にファイルが置かれますが、、、
gzipされている上に拡張子がない、しかも改行もない。
参考までにこんな感じです

```
{"messageType":"DATA_MESSAGE","owner":"xxx","logGroup":"redshift-experiment","logStream":"command-stream","subscriptionFilters":["All"],"logEvents":[{"id":"xxx","timestamp":1557395415730,"message":"hello1"}]}{"messageType":"DATA_MESSAGE","owner":"xxx","logGroup":"redshift-experiment","logStream":"command-stream","subscriptionFilters":["All"],"logEvents":[{"id":"xxx","timestamp":1557395420146,"message":"hello2"}]}
```

というわけでlambdaで変換します

```
import base64
import gzip
import io
import json

def lambda_handler(event, context):
    records = [process_record(r) for r in event['records']]
    return {'records': records }

def process_record(record):
    record_id = record['recordId']
    data = base64.b64decode(record['data'])
    iodata = io.BytesIO(data)

    with gzip.GzipFile(fileobj=iodata, mode='r') as f:
        data = json.loads(f.read())

    processed_data = process_data(data) + '\n'
    return {
        'data': base64.b64encode(processed_data.encode('utf-8')).decode('utf-8'),
        'result': 'Ok',
        'recordId': record_id
    }

def process_data(data):
    return '\n'.join([format_log_event(json) for json in data['logEvents']])

def format_log_event(j):
    return json.dumps({'timestamp': j['timestamp'], 'message': j['message']})
```

こんな感じのscriptをAWS Lambda create functionで適当な名前をつけて保存してあげて、、
あとはfirehoseのeditのTransform source records with AWS Lambdaで保存したlambdaを指定すれば変換されます。

lambda設定するとIAM roleでおこられるのでCreate new or updateクリック

これでいい感じのファイルがs3に置かれるのでAthena使えます。

```
{"timestamp": 1557401834708, "message": "hello"}
{"timestamp": 1557401837275, "message": "hello"}
```

Redshift使いたい場合は [こちら](https://qiita.com/takayuki_tk/items/20cb843df803e242bf92)のSourceを最初につくったKinesis StreamにすればOK

参考サイト
https://dev.classmethod.jp/cloud/aws/put-cloudwatchlogs/
https://qiita.com/tilfin/items/150cd899077d3c6b7edd