S3イベントでECSを起動する【前編】

# はじめに
ユーザがS3にファイルをアップロードした際にそのS3イベントをトリガーとして何らかの変換処理を実施し、DWH/DataLakeに格納する、といった簡易システムのニーズは多い。
今回はLambdaでは処理出来ないような重い処理、かつ不定期な利用頻度である場合にECS Fargateを用いてコスパ良く対処するアーキテクチャの開発を実施するので、備忘として記録する。
今回は前編として、S3 Object CreatedをトリガーとしてECSでコンテナを起動する所までを実装する。[後編](https://qiita.com/bokuikkun/items/1885662f401433914771)では、コンテナ内のpythonを用いてRDS for MySQLへのインサートを実装する。

# 概要 
![スクリーンショット 2023-10-06 13.42.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/3ca83eac-aa7b-9d1a-13f0-bb07901f3d5e.png)
このように、S3のObject CreatedイベントからEventBridgeを噛んでStepFunctionsを起動する。ECSはFargateで済む様であればその方がよく、EC2を使う必要があれば状況に応じて変更すれば良い。
本稿では、***バケットへのダミーcsvファイルのアップロードをトリガーにRDSに変換済みデータを格納する***処理を実装する。ユーザへのロール付与やUIの実装などはスコープ外とした。
また、S3イベント→Amazon EventBridgeの呼び出し確認としてSNSでのE-mail通知も要件外であるが、おまけで実装する。

# 開発環境
EC2にLinux環境(t3.midium & EBS32G)を立ち上げてコンテナイメージの作成を行った。

# 環境構築手順
1. [S3バケットの作成](#s3バケットの作成)
1. [SNSトピックの作成](#snsトピックの作成)
1. [Amazon EventBridgeルールの作成](#amazon-eventbridgeルールの作成)
1. [ECRへのイメージ登録](#ecrへのイメージ登録)
1. [AWS Batchの設定](#aws-batchの設定)
1. [StepFunctionsの設定](#stepfunctionsの設定)
1. [作成したEventBridgeへルールの追加](#作成したeventbridgeへルールの追加)
1. [デモンストレーション](#デモンストレーション)

## S3バケットの作成
適当なリージョンにバケットを作成し、作成後プロパティ→Amazon EventBridge→編集→通知を送信を"オン"に設定する。
![スクリーンショット 2023-10-02 11.45.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/c6f0338a-7b55-dfec-789d-ccc4e461a889.png)

## SNSトピックの作成
Amazon SNSからスタンダードタイプでトピックを作成します。
![68747470733a2f2f71696974612d696d6167652d73746f72652e73332e61702d6e6f727468656173742d312e616d617a6f6e6177732e636f6d2f302f3430393738332f63333136616434632d623932632d376638362d303463352d3730393736636161383735362e706e67.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/baf6ee03-bb23-ab5e-7673-be2ebc3f0b38.png)
今回は開発途中で私だけが通知を受け取れるようにしたいだけなので、アクセスポリシー→ベーシックタイプを選択→発行者と購読者に自身のAWSアカウントを指定、と設定を進めてトピックを作成します。
![スクリーンショット 2023-10-06 15.11.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/d0d7991d-4635-6b4c-7252-346f11e1ac60.png)
トピックが作成されたら、詳細設定からサブスクリプションの作成を行う。ARNには作成したトピックが示されています。
![スクリーンショット 2023-10-06 15.16.13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/b03edfed-1210-e7d0-ef95-a55f528d50a8.png)
すると、下記の様な購読確認メールが届くので、購読の確認を押下する。
![スクリーンショット 2023-10-09 10.18.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/8899259f-e9b0-557d-0833-5fc62122685d.png)
再度、SNSトピック側でサブスクリプションの詳細を確認するとステータスが確認済みになっていることが分かる。
![スクリーンショット 2023-10-09 11.44.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/49b6e30d-5d44-b4a2-e2bc-cfe63a519bdc.png)

## Amazon EventBridgeルールの作成
Amazon EventBridge→ルールを作成、からイベントに基づくルールを作成する。
![スクリーンショット 2023-10-09 12.14.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/4289c97e-27a3-e540-d54f-e06e5d3cd7fc.png)
イベントパターンでは、S3 Object Createdを指定する。バケット名を指定できるので、[S3バケットの作成](##-S3バケットの作成)の節で作成したバケット名を入力する。(プリフィックスを指定する際はjsonを適宜変更)
![スクリーンショット 2023-10-09 12.27.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/d3d10b80-0bc6-31e0-8ed3-33719e754950.png)
最後にイベントのターゲットにSNSトピックを指定する。
![スクリーンショット 2023-10-09 12.40.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/ce695149-9b1d-6d7e-03e6-60d9d08c6bca.png)
ここで、実際にS3に適当なファイルをアップロードするとEメール通知が届くことが確認できた。
![68747470733a2f2f71696974612d696d6167652d73746f72652e73332e61702d6e6f727468656173742d312e616d617a6f6e6177732e636f6d2f302f3430393738332f62663037383735312d306562342d303838382d633338632d6335613065356565373430622e706e67.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/2c7da5f5-ffe1-e083-2761-97959f47d30d.png)

## ECRへのイメージ登録
前述の通り、EC2のLinux環境を使ってDockerイメージを作成し、ECRへリポジトリ登録を実施していく。
今回はS3 Create Object→ストレージに保存するコンテナ(Python3)とストレージに保存したファイル→RDSに格納するコンテナ(SQL)の2つを作成する。
まずはPython3の方のDockerイメージの作成&ECR登録について説明した後、同じ手順に沿ってSQL側も作成していく(説明略)。
ディレクトリ構造は下記の通り。ただし、(※)付のファイルは認証情報をコンテナに引き渡すためのファイル。詳細は[こちら](https://qiita.com/bokuikkun/items/ad5e9ea7c1512c85b400)。
```
.
└── home
    └── ec2-user
        ├── Dockerfile
        ├── env.list 
        ├── docker-compose.yml (※)
        ├── .env (※)
        └── input
            ├── main.py
            ├── requirements.txt
            └── run.sh
```

### Pythonファイルの作成
今回はconvert処理内容は問わないので、とりあえずReadしたデータをそのまま掃き出すコードを記述する。データセットはIris.csvとし、名前をinput.csvにして保存しておく(後述のシェルスクリプトにて処理)。
```python:main.py
import pandas as pd



def read_csv(filename='../input.csv'):
    df = pd.read_csv(filename)
    return df


def convert(df):
    return df


def save(df, outfilename='../output.csv'):
    df.to_csv(outfilename)


def main():
    df = read_csv()
    converted = convert(df)
    save(converted)


if __name__=='__main__':
    main()    
```

### requirements.txtの作成
このファイルはpythonファイルの実行にインストールが必要な外部ライブラリをリスト化した物である。後のrun.sh内にてpip3による一括インストールを実施する。例として、取り敢えずnumpyとpandasを指定
```requirements.txt
numpy
pandas
```

### シェルスクリプトの作成
Dockerfileでこのシェルを実行する事でタスクの挙動を設計する。初めに、S3からアップロードされたinput.csvをコピーし、先ほどのpythonファイルを実行させるコマンドを記述する。
ここで、バケット名(BUCKET_NAME)とオブジェクトキー(OBJECT_KEY)を環境変数として記述してあるが、これは開発時にはコンテナを実行する際に環境変数の受け渡しを行い、本番環境ではStepFunctionsで指定する。
```shell:run.sh
#!/bin/sh
aws s3 cp s3://$BUCKET_NAME/$OBJECT_KEY /home/input.csv
python3 /home/input/main.py
```
### Dockerfileの作成
```Dockerfile: Dockerfile
### use Amazon Linux 2-based Docker images
FROM amazonlinux:2
USER root

### install the required packages
RUN yum -y update && \
    yum -y install python3-pip && \
    pip3 install awscli

### start bash shell when container is started
COPY input/ /home/input/
RUN pip3 install --user -r /home/input/requirements.txt
CMD ["sh", "/home/input/run.sh"]
```

### Dockerイメージの作成
ここで、開発環境下でビルドと実行確認を行う。
```zsh
$ docker build . -t aws-cli-container
```
これでaws-cli-containerというDockerイメージが作成される。
続いて開発環境であるEC2にS3にアクセスするIAMロールをアタッチする。AWSコンソールからIAM→ロールの作成→エンティティでAWSサービス→ユースケースでEC2を選択、と進みAmazonS3ReadOnlyAccessポリシーを選択する。今回はAllowEC2toReadS3という名前でロールを作成した。
![スクリーンショット 2023-10-10 10.52.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/30ece18d-11dd-f3d9-1b7b-e64168aaad28.png)
次にEC2のコンソールから開発環境のインスタンスにチェックを入れ、アクション→セキュリティ→IAMロールを変更、で作成したIAMロールをアタッチする。
![スクリーンショット 2023-10-10 11.01.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/23cc6019-8961-5a2b-5102-4c8400e7dc67.png)
ここで、テスト実行を行いたいので実際にS3バケットにinput.csv(元のIris.csv)をアップロードし、Dockerfileと同じディレクトリに環境変数ファイルを作成する。
```yml:env.list
BUCKET_NAME=#S3バケット名
OBJECT_KEY=input.csv
```
この上で、```--env-file```オプションを使ってコンテナを実行する。
```zsh
$ docker run -it --rm --env-file env.list aws-cli-container
```
[実行結果]
![スクリーンショット 2023-10-11 13.14.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/af1b6c76-8e0c-17a7-89c0-cfbab6c357c1.png)
この様に、```aws s3 cp```が正常に動作していれば良い。
実際にDockerfileに```bash```コマンドの実行も追加して、```home/input/```を見に行くとpythonファイルが掃き出した```output.csv```が作成されている事も確認できる。

先に概要だけ述べているが、***EC2上でコンテナへのawscliの利用は[こちらの記事](https://qiita.com/bokuikkun/items/ad5e9ea7c1512c85b400)にまとめている***ので、参考にして頂ければと思います。

## ECRへのDocker image登録
前章にてEC2環境下で作成したDockerイメージをECRに登録する。
初めに、Elastic Container Registryでリポジトリを作成する。Amazon Elastic Container Registryのコンソール画面からリポジトリの作成を押下。特に理由はないが、プライベートリポジトリを選択し、適当な名前を設定して作成
![スクリーンショット 2023-10-11 13.31.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/00d158a0-05a2-f31a-0d4b-dbcf5b7fb757.png)
リポジトリが作成されると詳細画面の右上からプッシュコマンドの表示、というボタンがあるのでこれを参考にしてイメージのプッシュを行う。
![スクリーンショット 2023-10-11 13.37.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/85aca421-e494-8147-535a-704fa03e3d55.png)
[プッシュコマンド]
![スクリーンショット 2023-10-11 13.39.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/15a9b2cc-df53-75b1-30d3-16b73d30afe5.png)
因みに、プッシュコマンドの手順を全て実施するとこの様な標準出力がなされる。
![スクリーンショット 2023-10-11 13.44.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/06653db9-1c7d-70cf-feae-42c3c7249ad1.png)

## AWS Batchの設定
続いて、リポジトリに登録したコンテナのスケーリングを制御するために、AWS Batchの設定を実施する。
本節はIAMロールの設定やネットワーク周りなど若干躓くポイントが多かったため、それぞれについて項を分けて記述する。
### コンピューティング環境の作成
コンピューティング環境の作成を行う。
AWS Batchのコンソール画面から、コンピューティング環境→作成と遷移する。
![スクリーンショット 2023-10-11 13.53.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/5b140dc5-eb4d-2368-0dd4-243b8dcfdaf1.png)
筆者が担う案件では、重い処理が単発かつ不定期に発生する想定なので、Fargate Spotでリソースを定義する。
![スクリーンショット 2023-10-11 13.56.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/0f12f0a8-5ebe-2555-ad33-3e044afbcc70.png)
### VPCおよびサブネットの設定
ここでネットワーク設定は、***ECSタスクにインターネットへのアウトバウンドが許可されていないと、Dockerイメージをpull出来ない点に留意***する。
![スクリーンショット 2023-10-12 14.40.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/1ab236f5-b008-b302-cb33-a51b1d10abba.png)
### ジョブキューの作成
ジョブキューの名前を設定し、先ほど作成したコンピューティング環境を選択して作成
![スクリーンショット 2023-10-11 14.03.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/5a5e3abd-a8ad-a271-de8d-6244572955aa.png)
### ジョブ定義の作成
最後にジョブ定義を作成する。
ジョブ定義の名前を設定し、オーケストレーションタイプにFargateを選択。
タスク実行ロール(Task Execution Role)には下記3つのポリシーを添付したロールをアタッチする。
- AmazonEC2ContainerRegistryReadOnly
- AmazonS3ReadOnlyAccess
- AmazonECSTaskExecutionRolePolicy (※)
![スクリーンショット 2023-10-12 14.26.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/1fdcd794-3de1-10ba-6f95-d243708b832d.png)
※AmazonECSTaskExecutionRolePolicyポリシーには```logs:CreateLogGroup```の許可が含まれないため、これをベースとしたカスタマー管理ポリシー```ECStaskExecutionRolePolicy```を作成した。
```json:ECStaskExecutionRolePolicy
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "ecr:GetAuthorizationToken",
                "ecr:BatchCheckLayerAvailability",
                "ecr:GetDownloadUrlForLayer",
                "ecr:BatchGetImage",
                "logs:CreateLogStream",
                "logs:CreateLogGroup",
                "logs:PutLogEvents"
            ],
            "Resource": "*"
        }
    ]
}
```
この様に作成したIAMロールをタスク実行ロールとして設定する。
![スクリーンショット 2023-10-12 14.27.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/cf68979e-4800-7d68-cfd9-6794fa3fde66.png)
次の画面に遷移すると、コンテナ設定に移るのでECRに登録したイメージのURIをコピー&ペーストする。
また、コマンドはデフォルトで```hello world```が出力されるようになっているが不要なので削除しても良い。
![スクリーンショット 2023-10-13 23.36.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/fee1f905-2441-0046-b99b-29a3301a5e19.png)
さらに画面下に遷移すると、ジョブロール(Task role)設定があるので、ここで利用するAWSリソースの権限を付けたIAMロールを選択する。今回のケースは下記の4つのポリシーを付けた。
- AmazonECS_FullAccess
- AmazonRDSFullAccess
- AmazonS3FullAccess
- AWSStepFunctionsFullAccess
![スクリーンショット 2023-10-12 14.36.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/61cb372f-3846-ff15-d0e6-85813f717582.png)
ロール名を```AllowECStaskstoAccessS3andRDSFully```とし、ジョブロールに設定する。
![スクリーンショット 2023-10-12 14.38.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/14138769-f737-a0e4-f817-5ad01bf28314.png)
## StepFunctionsの設定
まずは、先ほどECRに登録したコンテナだけを動かすワークフローを作成する。
今後、同ワークフローにMySQLによる.csv→RDSの格納やユーザへのSNS通知、エラー処理などを組み込むことを想定している。
StepFunctionsのコンソール画面から、ステートマシンの作成→Blankを選択してSelectを押下
![スクリーンショット 2023-10-11 15.07.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/43b46a92-94b0-37f8-9d2b-008c2644556b.png)
ワークフロー設計画面左側の検索窓から"AWS Batch SubmitJob"を選択&ワークフローに追加。状態名、ジョブ名を入力し、前章で作成したジョブ定義、ジョブキューを設定する。
![スクリーンショット 2023-10-11 15.19.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/02008546-3878-c6b5-151d-de6a1e54c1d4.png)
次にワークフロー設計画面から上部ボタンでデザイン→Codeに切り替えるとjson形式での定義が編集できる。ここで"Parameters"の下に"ContainerOverrides"を追記する事でコンテナ内のバケット名、オブジェクト名を参照可能に出来る。
![スクリーンショット 2023-10-11 15.26.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/f0f961e7-8a75-c8f8-9bef-990bf6d62abb.png)

## 作成したEventBridgeへルールの追加
先に作成した同じEventBridgeのルールのターゲットにStepFunctionsのステートマシンを追加する。別のターゲットを追加、というボタンを押下。ルールを更新する。
![スクリーンショット 2023-10-11 15.55.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/1348882c-cad8-aa27-9da9-b22cdb13c485.png)

## デモンストレーション
さて、ここまで実装出来たところで実際にS3に.csvファイルをアップロードしてみて一連のフローが正しく動作するか確認する。
まずは、S3にinput.csvをアップロード
![スクリーンショット 2023-10-13 7.50.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/305aee9e-5623-b01d-6558-ce8059a8a828.png)
S3へのObject CreatedイベントはEventBridgeを介して、SNS通知とStepFunctionsへ配信されるはずなので、先ほど確認した様なSNS通知が来ていればトリガーは成功しており、StepFunctionsへもイベントがpushされている筈である。
次にStepFunctionsのコンソール画面から作成したステートマシンの詳細を確認すると、このように過去のジョブの実行状況が確認できる。
![スクリーンショット 2023-10-13 7.51.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/05c9c146-8bdc-bd7e-8361-68234c942f19.png)
さらにこの実行名をクリックすると、エラーが発生した場合にどこのフローで失敗したかJSON形式で詳細を確認する事ができる。
成功の場合は下記の通りになる。
![スクリーンショット 2023-10-13 7.54.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/409783/2ab4445d-f48d-082f-0a7b-cd6ee72de355.png)

# おわりに
今回はS3 Object CreatedをトリガーとしてECS Farget Spotを起動するシステムの構築法を記した。
次回の[後編](https://qiita.com/bokuikkun/items/1885662f401433914771)では、ECSのコンテナ内のpythonでMySQLを叩いてRDSに格納するフローを実装する。