AWS Step FunctionsとLambdaでCloudWatch LogsのログをS3に定期的にエクスポートする

CloudWatch LogsのログをS3にエクスポートする方法としてはKinesis Firehoseなどがありますが、頻繁にエクスポートしなくても良い場合もあります。
その場合の選択肢の1つとしてStep Functionsもあるのかなと思って実装してみました。

## 大まかなながれ

* CloudWatch Eventsで定期的にStep Functionsを実行
* Step Functionsで複数のログに対してLambdaを実行
* CloudWatch LogsのS3エクスポートタスクを実行
    * ロググループ名やバケット名をStep Functionsから渡すことによってLambdaを汎化できる
* CloudWatch LogsのS3エクスポートタスクをポーリングする
    * 複数同時にS3へのエクスポートタスクを実行できないため

![stepfunctions.png](https://qiita-image-store.s3.amazonaws.com/0/45760/144d2529-9121-6373-d352-64d2a418f15f.png)


## Step Functions

### おさらい

* 視覚的なワークフローを使用して、分散アプリケーションとマイクロサービスのコンポーネントを調整できるウェブサービス
* API Gateway、CloudWatch Eventsからキックすることができる
* タスクとしてLambdaや（EC2、ECS）を実行する
* Step Functions自体はサーバレス
* リトライ制御が充実していて、失敗時に徐々に待ち時間を伸ばすことができる
* 課金体系は状態遷移する毎に課金
* JSONでStepを書く必要がある
* ifとかforとかは無いので戸惑う
    * 言語仕様
    * https://states-language.net/spec.html
* ステート間で多少のデータの受け渡しは可能

### ステート

| States | 説明 |
|:--|:--|
| Pass | 入力や出力を上書きして次のStateを実行することができる |
| Task | Lambdaやアクティビティ（EC2、ECS）などを実行できる<br/>タイムアウトやリトライを設定することができる |
| Choice | switch case文的なもの<br/>条件にマッチした場合とデフォルトの次のStateを指定できる |
| Wait | 秒数、指定したTimestampまで待つ<br/>Inputから受け取ったJSON内の秒数、Timestampを利用することも可能 |
| Succeed | 成功して終わる |
| Fail | 失敗して終わる |
| Parallel | 並列にStateを実行できる<br/>今回使っていないのでOutputがどうなるのかは未検証 |

## 前提条件
* 2018/07/19時点での内容です
* AWSのリージョン
    * ap-northeast-1
* CloudWatch Logsの複数のロググループを1日1回S3へエクスポートする
* CloudWatch Logs
    * ログがすでに出力されている
        * /test/log-group-a
        * /test/log-group-b
        * /test/log-group-c
* 出力先のS3
    * バケットがすでに用意されている
    * stepfunction-sample
* Lambda
    * ランタイム：Node.js8.10

## 手順

### IAM Policy、信頼関係、Roleを作成する

#### LambdaがCloudWatch Logsを操作するためのRole

<details><summary>Policy</summary>
<div>

```json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:DescribeExportTasks"
            ],
            "Resource": "arn:aws:logs:ap-northeast-1:123456789012:log-group:*"
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:ap-northeast-1:123456789012:log-group:/aws/lambda/cwl-to-s3-check-task:*",
                "arn:aws:logs:ap-northeast-1:123456789012:log-group:/aws/lambda/cwl-to-s3-create-task:*"
            ]
        },
        {
            "Sid": "VisualEditor2",
            "Effect": "Allow",
            "Action": [
                "logs:CreateExportTask"
            ],
            "Resource": [
                "arn:aws:logs:ap-northeast-1:123456789012:log-group:/test/log-group-a*",
                "arn:aws:logs:ap-northeast-1:123456789012:log-group:/test/log-group-b*",
                "arn:aws:logs:ap-northeast-1:123456789012:log-group:/test/log-group-c*"
            ]
        }
    ]
}
```

</div>
</details>

<details><summary>信頼関係</summary>
<div>

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
```

</div>
</details>

ロールARN：arn:aws:iam::123456789012:role/role-cwl-lambda

#### Step FunctionsがLambdaを実行するためのRole

<details><summary>Policy</summary>
<div>

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": [
                "*"
            ]
        }
    ]
}
```

</div>
</details>

<details><summary>信頼関係</summary>
<div>

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "Service": "states.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
```

</div>
</details>

ロールARN： arn:aws:iam::123456789012:role/role-cwl-step-function

### S3のバケットポリシーを設定

<details><summary>バケットポリシー</summary>
<div>

```json
{
    "Version": "2012-10-17",
    "Id": "PolicyID",
    "Statement": [
        {
            "Sid": "StmtID2",
            "Effect": "Allow",
            "Principal": {
                "Service": "logs.ap-northeast-1.amazonaws.com"
            },
            "Action": "s3:GetBucketAcl",
            "Resource": "arn:aws:s3:::stepfunction-sample"
        },
        {
            "Sid": "StmtID3",
            "Effect": "Allow",
            "Principal": {
                "Service": "logs.ap-northeast-1.amazonaws.com"
            },
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::stepfunction-sample/*",
            "Condition": {
                "StringEquals": {
                    "s3:x-amz-acl": "bucket-owner-full-control"
                }
            }
        }
    ]
}
```

</div>
</details>

### Lambda関数を作成する

#### CloudWatch LogsからS3へのエクスポートタスクを作成する関数

* 名前
    * cwl-to-s3-create-task
* ロールARN
    * arn:aws:iam::123456789012:role/role-cwl-lambda


<details><summary>ソースコード</summary>
<div>

```javascript
const AWS = require('aws-sdk');
const cwl = new AWS.CloudWatchLogs({apiVersion: '2014-03-28'});

exports.handler = async function handler(event, context) {
  const today = new Date();
  const from = new Date(today.getFullYear(), today.getMonth(), today.getDate() - 1, 0, 0, 0, 0);
  const to = new Date(today.getFullYear(), today.getMonth(), today.getDate() - 1, 23, 59, 59, 999);
  const s3BucketName = event.s3BucketName;
  const logGroupName = event.logGroupName;
  const destPrefix = event.destPrefix;

  try {
    const ret = await exportTask(s3BucketName, destPrefix, logGroupName, from, to);
    context.succeed({
      s3BucketName: s3BucketName,
      destPrefix: destPrefix,
      logGroupName: logGroupName,
      taskId: ret.taskId
    });
  } catch (e) {
    if (e.code == 'ResourceNotFoundException') {
      context.succeed({
        s3BucketName: s3BucketName,
        destPrefix: destPrefix,
        logGroupName: logGroupName,
        taskId: event.taskId
      });
    } else {
      context.fail(e);
    }
  }
}

async function exportTask(s3BucketName, destPrefix, logGroupName, from, to) {
  const date = from.toLocaleDateString('ja-JP', {year: 'numeric', month: '2-digit', day: '2-digit'});
  const destinationPrefix = destPrefix + logGroupName + '/' + date;
  const params = {
    destination: s3BucketName,
    from: from.valueOf(),
    logGroupName: logGroupName,
    to: to.valueOf(),
    destinationPrefix: destinationPrefix
  };
  return await cwl.createExportTask(params).promise();
}
```

</div>
</details>

#### CloudWatch LogsからS3へのエクスポートタスクのステータスを確認してハンドリングする

* 名前
    * cwl-to-s3-check-task
* ロールARN
    * arn:aws:iam::123456789012:role/role-cwl-lambda


<details><summary>ソースコード</summary>
<div>

```javascript
const AWS = require('aws-sdk');
const cwl = new AWS.CloudWatchLogs({apiVersion: '2014-03-28'});

exports.handler = handler;

class CheckError extends Error {
  constructor(name) {
    super(name);
    this.name = name;
  }
}

async function handler(event, context) {
  const s3BucketName = event.s3BucketName;
  const logGroupName = event.logGroupName;
  const destPrefix = event.destPrefix;
  const taskId = event.taskId;
  try {
    const ret = await describeTask(taskId);
    const retTask = ret.exportTasks[0];
    const statusCode = retTask.status.code;

    if (statusCode == 'COMPLETED') {
      context.succeed({
        s3BucketName: s3BucketName,
        logGroupName: logGroupName,
        destPrefix: destPrefix,
        taskId: taskId,
        status: retTask.status.code
      });
    } else {
      context.fail(new CheckError(statusCode));
      return;
    }
  } catch (e) {
    context.fail(e);
  }
}

async function describeTask(taskId) {
  const params = {
    taskId: taskId
  };
  return await cwl.describeExportTasks(params).promise();
}
```

</div>
</details>

### Step Functionsを作成する

* ステートマシンの名前
    * cwl-to-s3
* IAMロールのARN
    * arn:aws:iam::123456789012:role/role-cwl-step-function

<details><summary>ステートマシンの定義</summary>
<div>

```json
{
    "StartAt": "FirstPass",
    "States": {
      "Succeed": {
        "Type": "Succeed"
      },
      "Fail": {
        "Type": "Fail"
      },
      "FirstPass": {
        "Type": "Pass",
        "Result": {
            "logGroupName": "",
            "s3BucketName": "stepfunction-sample",
            "destPrefix": "path-prefix"
          },
          "Next": "Dispatch"
      },
      "Dispatch": {
        "Type": "Choice",
        "Choices": [
          {
            "Variable": "$.logGroupName",
            "StringEquals": "",
            "Next": "log-group-a"
          },
          {
            "Variable": "$.logGroupName",
            "StringEquals": "/test/log-group-a",
            "Next": "log-group-b"
          },
          {
            "Variable": "$.logGroupName",
            "StringEquals": "/test/log-group-b",
            "Next": "log-group-c"
          },
          {
            "Variable": "$.logGroupName",
            "StringEquals": "/test/log-group-c",
            "Next": "Succeed"
          }
        ],
        "Default": "Fail"
      },
      "log-group-a": {
        "Type": "Pass",
        "Result": "/test/log-group-a",
        "ResultPath": "$.logGroupName",
        "Next": "CwlCreateTask"
      },
      "log-group-b": {
        "Type": "Pass",
        "Result": "/test/log-group-b",
        "ResultPath": "$.logGroupName",
        "Next": "CwlCreateTask"
      },
      "log-group-c": {
        "Type": "Pass",
        "Result": "/test/log-group-c",
        "ResultPath": "$.logGroupName",
        "Next": "CwlCreateTask"
      },
      "CwlCreateTask": {
        "Type": "Task",
        "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:cwl-to-s3-create-task",
        "Catch": [
          {
            "ErrorEquals": ["ResourceNotFoundException"],
            "Next": "Dispatch"
          },
          {
            "ErrorEquals": ["States.ALL"],
            "Next": "Fail"
          }
        ],
	"Next": "CwlCheckTask"
      },
      "CwlCheckTask": {
        "Type": "Task",
        "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:cwl-to-s3-check-task",
        "Retry": [
            {
                "ErrorEquals": [ "RUNNING", "PENDING" ],
                "IntervalSeconds": 60,
                "BackoffRate": 2.0,
                "MaxAttempts": 3
            }
        ],
        "Catch": [
            {
                "ErrorEquals": [ "States.ALL" ],
                "Next": "Fail"
            }
        ],
        "Next": "Dispatch"
      }
    }
  }
```
</div>
</details>

### CloudWatch EventsでStep Functionsを定期的にキックする

* イベントソース：スケジュール
    * Cron式：0 1 * * ? *
        * GTM時なので注意が必要
* ターゲット
    * Step Functions ステートマシン
* ステートマシン
    * cwl-to-s3
* role
    * この特定のリソースに対して新しいロールを作成する
    * arn:aws:iam::123456789012:role/service-role/role-event-cwl-to-s3
* ルールの定義
    * 名前
        * event-rule-cwl-to-s3
    * 状態
        * 有効

## 参考文献
 * https://docs.aws.amazon.com/ja_jp/AmazonCloudWatch/latest/logs/S3Export.html
 * https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/welcome.html
 * https://dev.classmethod.jp/cloud/aws/step-functions-from-cron/