docker-composeで立てたコンテナ内のDigdagからLambda(API Gateway)を実行させてみた。

# この記事の内容
![http://docs.digdag.io/logo.html#horizontal](http://docs.digdag.io/_static/logo/logo-digdag-rec-tr.png)
これはまだ構想段階ではありますが、将来ECSで立てたコンテナ内のDigdagから、Lambdaをキックさせたいです。

その動作検証の意味で、まずはdocker-composeを使ってローカル環境にDigdagサーバーとPostgreSQLサーバーを立てて、その後ローカルのDigdagからLambda(API Gateway)を実行させました。

当記事はその備忘録です。

※ Lambdaの作成手順・API Gatewayの設定手順は記載しておりません。API Gatewayのエンドポイントへリクエストを送るとLambdaがキックされる状態であることを前提に話を進めております。

# 備忘録
## ローカル環境構築
ローカルのディレクトリ構成ですが以下の通りです。
※ docker-compose実行後に「./digdag/bin/digdag」が生成されるはずです。

```
$ tree
.
├── digdag
│   ├── Dockerfile
│   ├── etc
│   │   └── digdag.properties
│   └── startup.sh
├── docker-compose.yml
├── postgres
│   └── Dockerfile
```

各ファイルの内容は以下の通りです。

- ./digdag/Dockerfile

```Dockerfile:Dockerfile
FROM openjdk:8

# パッケージ管理システムのアップデート
RUN apt-get -y update && apt-get -y upgrade

# localeの設定
RUN apt-get -y install locales && \
    localedef -f UTF-8 -i ja_JP ja_JP.UTF-8
ENV LANG ja_JP.UTF-8
ENV LANGUAGE ja_JP:ja
ENV LC_ALL ja_JP.UTF-8

# timezone (Asia/Tokyo)の設定
ENV TZ JST-9

# 必要なものインストール
RUN apt-get install -y vim postgresql-client

WORKDIR /root

```

- ./digdag/etc/digdag.properties

```properties:digdag.properties
database.type = postgresql
database.user = fuka_user
database.password = fuka_pass
database.host = postgres_etl
database.port = 5432
database.database = digdag
```

- ./digdag/startup.sh

```shell:startup.sh
#!/bin/sh

# digdagをインストール(http://docs.digdag.io/getting_started.html)
curl -o ~/bin/digdag --create-dirs -L "https://dl.digdag.io/digdag-latest"
chmod +x ~/bin/digdag
echo 'export PATH="$HOME/bin:$PATH"' >> ~/.bashrc
. ~/.bashrc

digdag server --config ./etc/digdag.properties --bind 0.0.0.0 --port 65432 --task-log /var/log/digdag/ --access-log /var/log/digdag
```

- ./postgres/Dockerfile

```Dockerfile:Dockerfile
FROM postgres:latest
```

- ./docker-compose.yml

```yaml:docker-compose.yml
version: '3'

services:
  dbserver:
    container_name: postgres_etl
    build: ./postgres
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - PGPASSWORD=${PGPASSWORD}
      - DATABASE_HOST=${DATABASE_HOST}
      - POSTGRES_DB=${POSTGRES_DB}
    ports:
      - 5432:5432
    tty: true
  digdag:
    container_name: digdag_etl
    build: ./digdag
    volumes:
      - ./digdag:/root
      # /tmpディレクトリがないとDigdagサーバーのホーム画面が起動しないので明示的にマウント。
      - /tmp:/tmp
    ports:
      - "0.0.0.0:65432:65432"
    entrypoint: >
      sh -c "/root/startup.sh && /bin/bash"
    tty: true

networks:
  etl_networks:
```

以下の通りdocker-composeを実行して、Digdagコンテナ・Digdagサーバー・PostgreSQLコンテナを起動します。

```
$ docker-compose build --no-cache

$ docker-compose up -d
```

## Digdagコンテナの確認
ブラウザからDigdagサーバーにアクセスすると以下のホーム画面が表示されます。
![スクリーンショット 2021-09-01 11.29.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371897/a20a4946-8b2e-a09c-5692-f4bb396a82da.png)


ターミナルからDigdagコンテナに接続して、
**1. ワークフローを作成**
**2. Digdagサーバーへプロジェクト名を指定してワークフローをPUSH**
**3. ワークフロー実行**
の手順で進めることもできますが、今回はブラウザから進めていきます。

まずは画面右上の「New project」ボタンを押します。
すると以下のプロジェクト画面になるので、ここで好きなプロジェクト名を記入し、ワークフローを追加→保存（Add file→Save）していきます。
![スクリーンショット 2021-09-01 11.38.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371897/24e3297d-d838-cc9d-d40b-475edb799609.png)

以下の通り、今回はLambdaに紐づいたAPI Gatewayのエンドポイントを実行する処理だけを追加（Add file）し、保存（Save）します。
![スクリーンショット 2021-09-01 12.00.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371897/99d21cf5-d216-20d4-f35f-57b98e3aabba.png)

## PostgreSQLコンテナの確認
以下のようにdigdag DB内にアクセスし、

```
root@5dddd03ba95a:/# psql -d digdag -U fuka_user -h localhost
psql (13.4 (Debian 13.4-1.pgdg100+1))
Type "help" for help.

digdag-#
digdag-#
digdag-#
```

テーブル一覧を確認します。

するとDigdag用のテーブル達がデフォルトで作成されていることがわかります。

```
digdag-# \z
                                         Access privileges
 Schema |            Name             |   Type   | Access privileges | Column privileges | Policies 
--------+-----------------------------+----------+-------------------+-------------------+----------
 public | delayed_session_attempts    | table    |                   |                   | 
 public | projects                    | table    |                   |                   | 
 public | projects_id_seq             | sequence |                   |                   | 
 public | queue_settings              | table    |                   |                   | 
 public | queue_settings_id_seq       | sequence |                   |                   | 
 public | queued_task_locks           | table    |                   |                   | 
 public | queued_task_locks_id_seq    | sequence |                   |                   | 
 public | queued_tasks                | table    |                   |                   | 
 public | queued_tasks_id_seq         | sequence |                   |                   | 
 public | queues                      | table    |                   |                   | 
 public | queues_id_seq               | sequence |                   |                   | 
 public | resuming_tasks              | table    |                   |                   | 
 public | resuming_tasks_id_seq       | sequence |                   |                   | 
 public | revision_archives           | table    |                   |                   | 
 public | revision_archives_id_seq    | sequence |                   |                   | 
 public | revisions                   | table    |                   |                   | 
 public | revisions_id_seq            | sequence |                   |                   | 
 public | schedules                   | table    |                   |                   | 
 public | schedules_id_seq            | sequence |                   |                   | 
 public | schema_migrations           | table    |                   |                   | 
 public | secrets                     | table    |                   |                   | 
 public | secrets_id_seq              | sequence |                   |                   | 
 public | session_attempts            | table    |                   |                   | 
 public | session_attempts_id_seq     | sequence |                   |                   | 
 public | session_monitors            | table    |                   |                   | 
 public | session_monitors_id_seq     | sequence |                   |                   | 
 public | sessions                    | table    |                   |                   | 
 public | sessions_id_seq             | sequence |                   |                   | 
 public | task_archives               | table    |                   |                   | 
 public | task_dependencies           | table    |                   |                   | 
 public | task_dependencies_id_seq    | sequence |                   |                   | 
 public | task_details                | table    |                   |                   | 
 public | task_state_details          | table    |                   |                   | 
 public | tasks                       | table    |                   |                   | 
 public | tasks_id_seq                | sequence |                   |                   | 
 public | workflow_configs            | table    |                   |                   | 
 public | workflow_configs_id_seq     | sequence |                   |                   | 
 public | workflow_definitions        | table    |                   |                   | 
 public | workflow_definitions_id_seq | sequence |                   |                   | 
(39 rows)
```

ちなみにDigdagで実行するワークフローは、queuesテーブルなどに一時的に入れられロックされることで、重複して実行されないようになっています。

## Digdagワークフローの実行（API Gatewayの実行）
以下の通り、先ほどDigdagホーム画面から作成したワークフローを「Workflows」より確認できます。
![スクリーンショット 2021-09-01 12.06.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371897/54391b71-7a5b-a4d3-0b08-8ae00546931b.png)


「main」を選択すると以下の画面となります。
![スクリーンショット 2021-09-01 12.07.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371897/803f0f61-2262-a21f-db43-8acfc7905371.png)

上記の右上のRUNボタンを押すと、ワークフローが実行され、Lambdaがキックされることになります。
※ 実行中にPostgreSQLのqueuesテーブルなどの中身を見てみると、実行中のワークフローがレコードとして登録されているので見てみると面白いかもしれません。

以下実行後のワークフローのAttemptの画面。
![スクリーンショット 2021-09-01 12.13.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371897/f81e54f2-60f8-e8ad-aa2d-5e6cc4fd5053.png)

また処理失敗などしてしまい、再度途中から実行したい時などは、このAttemptという単位で途中から実行することができます。（実行履歴がPostgres内のテーブルに入っているため可能）

※ PostgreSQLでなくメモリでDigdagを起動している場合には多分不可能。