AWS Copilot + Embulk + RDSのスナップショット で個人情報をマスクしたDBを簡単に立ち上げられるようにする

はじめに

こんにちは。レンティオ株式会社でエンジニアをしている松田です。

突然ですが、皆さんは開発用のDBはどのように用意していますか？
できるだけ実運用に近いデータがあれば便利ですが、本番データをそのままコピーするわけにはいかないですし、かといっていろいろなパターンのseedデータを作るのもたいへんですよね。
そこで、レンティオでは、個人情報をマスクしたDBをRDSインスタンスをとして立ち上げられたり、ローカルDBにインポートできる環境を構築しました。
今回はそのしくみについて紹介できればと思います。

やっていることはそこまで複雑なわけではなく、

1. 本番DBのスナップショットから作業用DBを作成
2. Embulkを使って作業用DBのデータのマスクを行う
3. マスクしたDBからスナップショットを作成する、また、ローカルにインポートできるようにS3にもpg_dumpする

といった感じになります。
シンプルではありますが、このようにRDSのスナップショット機能を使うことで、

• 本番DBには直接接続しないのでパフォーマンスへの影響がない
• DBのサイズが大きくなってもスケールする

など、いろいろメリットがあります。

次に構成について詳しく見ていきます。

構成

本構成は、AWS Copilotを使って構築しています。
AWS Copilotは文字通りAWSが開発・管理しているツールキットで、Webサービスやスケジュールジョブなどのよく使われている環境を簡単に構築できます。
レンティオでは、さまざまなサービスの環境構築に採用していて、 過去に記事も出しているので気になった方はぜひ読んでみてください。
また、AWS CDKやTerraformなどAWS上でリソースを作成できるツールであれば同じことができますので適宜参考にしていただけばと思います。

マスク処理などのデータ加工にはEmbulkを使用しています。
Embulkはオープンソースのバルクデータローダで、プラグインを追加することでさまざまなデータ加工ができたり、いろいろなデータソースに対応できます。
また、JavaのほかにJRubyでも実行ができ、レンティオではJRubyで実行しています。

全体のディレクトリ構成は以下のようになっています。
AWS CopilotとEmbulkがメインで構成されていて、後は実行用のシェルファイルとDockerイメージがあります。
それぞれ個別に説明していきたいと思います。

.
├── .embulk/
│   └── embulk.properties
├── bin/
│   ├── copy_and_mask_local_db.sh
│   ├── create_masked_snapshot.sh
│   └── exec_embulk.sh
├── copilot/
│   ├── create-masked-snapshot/
│   │   ├── addons/
│   │   │   └── create-masked-snapshot.yml
│   │   └── manifest.yml
│   ├── environments/
│   │   ├── addons/
│   │   │   ├── addons.parameters.yml
│   │   │   └── template.yml
│   │   └── production-data/
│   │       └── manifest.yml
│   └── pipelines/
│       └── pipeline-seedgen-production-data/
│           ├── buildspec.yml
│           └── manifest.yml
├── embulk_bundle/
│   ├── .bundle/
│   │   └── config
│   ├── embulk/
│   │   ├── filter/
│   │   │   └── example.rb
│   │   ├── input/
│   │   │   └── example.rb
│   │   └── output/
│   │       └── example.rb
│   ├── Gemfile
│   └── Gemfile.lock
├── embulk_config/
│   ├── include/
│   │   ├── _in.yml.liquid
│   │   └── _out.yml.liquid
│   └── ...<your_tables>.liquid
├── embulk_plugins/
│   └── embulk-filter-anonymize/
│       ├── lib/
│       │   └── embulk/
│       │       └── filter/
│       │           └── anonymize.rb
│       ├── .ruby-version
│       ├── Gemfile
│       ├── Rakefile
│       └── embulk-filter-anonymize.gemspec
├── Dockerfile
├── Dockerfile.dev
└── compose.yml

AWS Copilot

アプリ名は create-masked-snapshot 、環境名は production-data 、 パイプライン名は pipeline-seedgen-production-data で作成しています。

.
└── copilot/
    ├── create-masked-snapshot/
    │   ├── addons/
    │   │   └── create-masked-snapshot.yml
    │   └── manifest.yml
    ├── environments/
    │   ├── addons/
    │   │   ├── addons.parameters.yml
    │   │   └── template.yml
    │   └── production-data/
    │       └── manifest.yml
    └── pipelines/
        └── pipeline-seedgen-production-data/
            ├── buildspec.yml
            └── manifest.yml

environmentsのマニフェストファイルですが、VPCとサブネットは本番環境のものを固定値でそのまま使っています。
本番環境と同じネットワーク内にリソースを作りたくない場合はVPCピアリングなどを使うと良いでしょう。

copilot/environments/production-data/manifest.yml

name: production-data
type: Environment

network:
  vpc:
    id: <vpc_of_your_application> 
    subnets:
      public:
        - id: <subnet_of_your_application>
        - id: <subnet_of_your_application>

observability:
  container_insights: false

また、Addon機能を使ってセキュリティグループとDBクラスタ用のパラメータグループを作成しています。
この2つのリソースは、作業用のDBを作成する際に使用します。

copilot/environments/addons/template.yml

Parameters:
  # テンプレートに定義する必要があるパラメータ
  App:
    Type: String
  Env:
    Type: String
  # 追加したパラメータ
  EnvironmentSecurityGroup:
    Type: String

Resources:
  SnapshotDBSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: 'Security group for RDS snapshot'
      VpcId: <vpc_of_your_application>
  SnapshotDBSecurityGroupIngress:
    Type: AWS::EC2::SecurityGroupIngress
    Properties:
      Description: !Sub 'Allow connect from ${App}-${Env}'
      GroupId: !Ref SnapshotDBSecurityGroup
      IpProtocol: tcp
      FromPort: 5432
      ToPort: 5432
      SourceSecurityGroupId: !Ref EnvironmentSecurityGroup
  DBClusterParameterGroupSnapshot:
    Type: AWS::RDS::DBClusterParameterGroup
    Properties:
      Description: 'Parameter group for the Snapshot RDS cluster'
      Family: aurora-postgresql16
      Parameters:
        timezone: "Asia/Tokyo"
        session_replication_role: replica

Outputs:
  SnapshotDBSecurityGroup:
    Value: !Ref SnapshotDBSecurityGroup
    Export:
      Name: !Sub '${App}-${Env}-SnapshotDBSecurityGroup'
  DBClusterParameterGroupSnapshot:
    Value: !Ref DBClusterParameterGroupSnapshot
    Export:
      Name: !Sub '${App}-${Env}-DBClusterParameterGroupSnapshot'

copilot/environments/addons/addons.parameters.yml

Parameters:
  EnvironmentSecurityGroup: !Ref EnvironmentSecurityGroup

アプリのマニフェストファイルですが、Scheduled Job で作成しています。
スナップショットを作成する際にDB名・ユーザー名・パスワードが必要になるので、SSMに保持しておきます。
このパスワードは本番とは別のものにしておいてください

copilot/create-masked-snapshot/manifest.yml

name: create-masked-snapshot
type: Scheduled Job

retries: 0
timeout: 5h

image:
  build: Dockerfile

cpu: 2048
memory: 4096
platform: linux/x86_64

secrets:
  SNAPSHOT_DB_DATABASE: /copilot/seedgen/${COPILOT_ENVIRONMENT_NAME}/secrets/snapshot_db_database
  SNAPSHOT_DB_USER: /copilot/seedgen/${COPILOT_ENVIRONMENT_NAME}/secrets/snapshot_db_user
  SNAPSHOT_DB_PASSWORD: /copilot/seedgen/${COPILOT_ENVIRONMENT_NAME}/secrets/snapshot_db_password

environments:
  production-data:
    on:
      schedule: "cron(0 20 L * ? *)" # 毎月1日の5時に実行

アプリ側にもAddon機能を使ってリソースポリシーを追加しておきます。
このポリシーは作業用のスクリプトを実行する際に使用します。
主なロールの内容としては、

• スナップショットの復元
• 作業用DBの作成・削除
• S3へのダンプ

などがあります。
DB削除のアクションについては、ワイルドカードを部分的に使ってできる限り狭い範囲で指定し、ほかのDBの削除はできないようにしておきましょう

copilot/create-masked-snapshot/addons/create-masked-snapshot.yml

Parameters:
  App:
    Type: String
    Description: Your application's name.
  Env:
    Type: String
    Description: The environment name your service, job, or workflow is being deployed to.
  Name:
    Type: String
    Description: Your workload's name.

Resources:
  createSnapshotPolicy:
    Metadata:
      'aws:copilot:description': 'An IAM ManagedPolicy for your service to create snapshots of the RDS'
    Type: AWS::IAM::ManagedPolicy
    Properties:
      Description: 'Grants access to the RDS snapshots'
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: SnapshotActions
            Effect: Allow
            Action:
              - rds:AddTagsToResource
              - rds:CreateDBClusterSnapshot
              - rds:CreateDBInstance
              - rds:DescribeDBClusterSnapshots
              - rds:DescribeDBClusters
              - rds:DescribeDBInstances
              - rds:RestoreDBClusterFromSnapshot
              - cloudformation:DescribeStacks
            Resource: "*"
          - Sid: DeleteProductionSnapshotClusterActions
            Effect: Allow
            Action:
              - rds:DeleteDBCluster
            Resource: "arn:aws:rds:ap-northeast-1:<your_aws_account_id>:cluster:snapshot-dbcluster-rds-*"
          - Sid: DeleteProductionSnapshotInstanceActions
            Effect: Allow
            Action:
              - rds:DeleteDBInstance
            Resource: "arn:aws:rds:ap-northeast-1:<your_aws_account_id>:db:snapshot-dbcluster-rds-*"
          - Sid: S3ObjectActions
            Effect: Allow
            Action:
              - s3:GetObject
              - s3:PutObject
              - s3:PutObjectACL
              - s3:PutObjectTagging
              - s3:DeleteObject
              - s3:RestoreObject
            Resource: "arn:aws:s3:::<your-bucket-name>/*"
          - Sid: S3ListAction
            Effect: Allow
            Action: s3:ListBucket
            Resource: "arn:aws:s3:::<your-bucket-name>"

Outputs:
  createSnapshotPolicy:
    Description: "The IAM::ManagedPolicy to attach to the task role"
    Value: !Ref createSnapshotPolicy

パイプラインのマニフェストファイルはほとんどデフォルトのままなので特に説明するところはなし。

copilot/pipelines/pipeline-seedgen-production-data/manifest.yml

name: pipeline-seedgen-production-data

version: 1

source:
  provider: GitHub
  properties:
    branch: main
    repository: https://github.com/<your_repogitory_path>

stages:
    name: production-data

buildspecファイルについても初回に自動生成されたものから特に変更はしていないです。
copilot linux binary のURLを最新のバージョンに適宜更新するくらいかなと思います。

copilot/pipelines/pipeline-seedgen-production-data/buildspec.yml

# Buildspec runs in the build stage of your pipeline.
version: 0.2
phases:
  install:
    commands:
      - echo "cd into $CODEBUILD_SRC_DIR"
      - cd $CODEBUILD_SRC_DIR
      # Download the copilot linux binary.
      - wget -q https://ecs-cli-v2-release.s3.amazonaws.com/copilot-linux-v1.34.1 -O copilot-linux
      - chmod +x ./copilot-linux
  build:
    commands:
      - echo "Run your tests"
      # - make test
  post_build:
    commands:
      - ls -l
      - export COLOR="false"
      - export CI="true"
      - pipeline=$(cat $CODEBUILD_SRC_DIR/copilot/pipelines/pipeline-seedgen-production-data/manifest.yml | ruby -ryaml -rjson -e 'puts JSON.pretty_generate(YAML.load(ARGF))')
      - pl_envs=$(echo $pipeline | jq -r '.stages[].name')
      # Find all the local services in the workspace.
      - svc_ls_result=$(./copilot-linux svc ls --local --json)
      - svc_list=$(echo $svc_ls_result | jq '.services')
      - >
        if [ ! "$svc_list" = null ]; then
          svcs=$(echo $svc_ls_result | jq -r '.services[].name');
        fi
      # Find all the local jobs in the workspace.
      - job_ls_result=$(./copilot-linux job ls --local --json)
      - job_list=$(echo $job_ls_result | jq '.jobs')
      - >
        if [ ! "$job_list" = null ]; then
          jobs=$(echo $job_ls_result | jq -r '.jobs[].name');
        fi
      # Raise error if no services or jobs are found.
      - >
        if [ "$svc_list" = null ] && [ "$job_list" = null ]; then
          echo "No services or jobs found for the pipeline to deploy. Please create at least one service or job and push the manifest to the remote." 1>&2;
          exit 1;
        fi
      # Generate the cloudformation templates.
      # The tag is the build ID but we replaced the colon ':' with a dash '-'.
      # We truncate the tag (from the front) to 128 characters, the limit for Docker tags
      # (https://docs.docker.com/engine/reference/commandline/tag/)
      # Check if the `svc package` commanded exited with a non-zero status. If so, echo error msg and exit.
      - >
        for env in $pl_envs; do
          tag=$(echo ${CODEBUILD_BUILD_ID##*:}-$env | sed 's/:/-/g' | rev | cut -c 1-128 | rev)
          for svc in $svcs; do
          ./copilot-linux svc package -n $svc -e $env --output-dir './infrastructure' --tag $tag --upload-assets;
          if [ $? -ne 0 ]; then
            echo "Cloudformation stack and config files were not generated. Please check build logs to see if there was a manifest validation error." 1>&2;
            exit 1;
          fi
          done;
          for job in $jobs; do
          ./copilot-linux job package -n $job -e $env --output-dir './infrastructure' --tag $tag --upload-assets;
          if [ $? -ne 0 ]; then
            echo "Cloudformation stack and config files were not generated. Please check build logs to see if there was a manifest validation error." 1>&2;
            exit 1;
          fi
          done;
        done;
      - ls -lah ./infrastructure
artifacts:
  files:
    - "infrastructure/*"

Dockerファイルと実行プログラム

実行用のプログラムはbinフォルダにシェルスクリプトとして書いており、Fargate(Dockerコンテナ)上で実行されます

.
├── bin/
│   ├── copy_and_mask_local_db.sh
│   ├── create_masked_snapshot.sh
│   └── exec_embulk.sh
├── Dockerfile
├── Dockerfile.dev
└── compose.yml

Dockerファイルでは、Embulkに必要なライブラリ・Gemのインストールを行った上でシェルスクリプトを実行しています。

Dockerfile

FROM --platform=linux/x86-64 your-image-path

WORKDIR /home

COPY embulk_config embulk_config/
COPY embulk_plugins embulk_plugins/
COPY embulk_bundle embulk_bundle/
COPY .embulk .embulk/
COPY bin bin/
RUN chmod -R a+rx embulk_config
RUN chmod -R a+rwx embulk_bundle
RUN chmod -R a+x embulk_plugins

RUN apk update \
    && apk add \
    aws-cli \
    openjdk17-jdk \
    git \
    curl \
    gzip \
    libc6-compat

RUN aws --version

# Gemfileのバージョンと合わせる
RUN curl -o embulk.jar -L https://github.com/embulk/embulk/releases/download/v0.11.4/embulk-0.11.4.jar && chmod +x embulk.jar

RUN curl -o jruby.jar -L https://repo1.maven.org/maven2/org/jruby/jruby-complete/9.4.7.0/jruby-complete-9.4.7.0.jar && chmod +x jruby.jar
RUN curl -o postgresql.jar -L https://jdbc.postgresql.org/download/postgresql-42.7.3.jar && chmod +x postgresql.jar

RUN java -jar embulk.jar gem install bundler -v 2.5.14
RUN cd embulk_bundle && java -jar ../embulk.jar bundle install

CMD bin/create_masked_snapshot.sh

Docker Compose用の設定ファイルもありますが、こちらはローカルでテスト実行するためのものになります。
ローカル環境のDockerを使ってアプリが立ち上がっていれば、そちらのDBへつなげるように設定しています。

compose.yml

services:
  seedgen:
    volumes:
      - ./bin:/home/bin
    build:
      context: .
      dockerfile: Dockerfile.dev
    extra_hosts:
      - host.docker.internal:host-gateway
    environment:
      - TZ=Asia/Tokyo
      - IN_PGHOST=host.docker.internal
      - IN_PGUSER=default
      - IN_PGPASSWORD=password
      - IN_PGDATABASE=rentio-development
      - OUT_PGHOST=host.docker.internal
      - OUT_PGUSER=default
      - OUT_PGPASSWORD=password
      - OUT_PGDATABASE=rentio-seedgen

ローカルでテスト実行する際のDockerファイルです。
本番用のものとほとんど同じですが、ローカルではスナップショットなどは使わないので、AWS CLI関連のライブラリはインストールしていないのと、実行するスクリプトが別になっています。

Dockerfile.dev

FROM --platform=linux/x86-64 your-image-path

WORKDIR /home

COPY embulk_config embulk_config/
COPY embulk_plugins embulk_plugins/
COPY embulk_bundle embulk_bundle/
COPY .embulk .embulk/
COPY bin bin/
RUN chmod -R a+rx embulk_config
RUN chmod -R a+rwx embulk_bundle
RUN chmod -R a+x embulk_plugins

RUN apk update \
    && apk add \
    openjdk17-jdk \
    git \
    curl \
    gzip \
    libc6-compat

# Gemfileのバージョンと合わせる
RUN curl -o embulk.jar -L https://github.com/embulk/embulk/releases/download/v0.11.4/embulk-0.11.4.jar && chmod +x embulk.jar

RUN curl -o jruby.jar -L https://repo1.maven.org/maven2/org/jruby/jruby-complete/9.4.7.0/jruby-complete-9.4.7.0.jar && chmod +x jruby.jar
RUN curl -o postgresql.jar -L https://jdbc.postgresql.org/download/postgresql-42.7.3.jar && chmod +x postgresql.jar

RUN java -jar embulk.jar gem install bundler -v 2.5.14
RUN cd embulk_bundle && java -jar ../embulk.jar bundle install

CMD bin/copy_and_mask_local_db.sh

こちらがスクリプトファイルになります。
stack_name=<your_stackname_of_copilot_env_addon> の箇所は、copilot env deployした際に作成されたAddon側のCloudFormationの名称を入れてください。

エラーやリトライの処理も入っていてかなり複雑に見えますが、主な処理の流れとしては、

1. 本番DBの最新のスナップショットを取得する
2. Copilot envのAddon機能を使って作成されたセキュリティグループとDBクラスタ用のパラメータグループを取得する
3. スナップショットを使って作業用DB(input, outputの2つ)を作成する
4. Embulk, postgres用の接続情報を設定する
5. 不要なDBをtruncateする
6. Embulkを実行する
7. DB名・ユーザー名・パスワードを本番のものから変更する
8. マスクしたDBのスナップショットを作成する
9. S3にダンプする
10. 作業用のDBを削除する

となります。

bin/create_masked_snapshot.sh

#!/bin/bash
set -e

cleanup() {
    echo "[error] Script failed. Cleaning up RDS resources..."

    # RDSインスタンスとクラスターを削除
    echo "[debug] Deleting RDS instances and clusters"
    aws rds delete-db-instance \
        --db-instance-identifier $db_instance_in \
        --skip-final-snapshot \
        --delete-automated-backups 2>/dev/null || true
    aws rds delete-db-instance \
        --db-instance-identifier $db_instance_out \
        --skip-final-snapshot \
        --delete-automated-backups 2>/dev/null || true
    aws rds delete-db-cluster \
        --db-cluster-identifier $db_cluster_in \
        --skip-final-snapshot \
        --delete-automated-backups 2>/dev/null || true
    aws rds delete-db-cluster \
        --db-cluster-identifier $db_cluster_out \
        --skip-final-snapshot \
        --delete-automated-backups 2>/dev/null || true

    echo "[debug] RDS resources have been cleaned up"
}

snapshot_db_cluster_identifier=<your_snapshot_db_cluster_identifier>
db_subnet_group_name=<your_db_subnet_group_name>

# 最新のスナップショットを取得する
snapshot=$(aws rds describe-db-cluster-snapshots --db-cluster-identifier $snapshot_db_cluster_identifier --query="max_by(DBClusterSnapshots, &SnapshotCreateTime).DBClusterSnapshotIdentifier" --output text)

# CopilotのenvスタックのAddon(作り直すと後半のIDが変わるので注意)
stack_name=<your_stackname_of_copilot_env_addon>

# Addonで作成したリソースを参照する
db_cluster_parameter_group_name=$(aws cloudformation describe-stacks --stack-name $stack_name --query="Stacks[].Outputs[?ExportName=='seedgen-production-data-DBClusterParameterGroupSnapshot'].[OutputValue]" --output text)
snapshot_db_security_group=$(aws cloudformation describe-stacks --stack-name $stack_name --query="Stacks[].Outputs[?ExportName=='seedgen-production-data-SnapshotDBSecurityGroup'].[OutputValue]" --output text)

echo "[debug] launch snapshot DB"

# 識別子の文字数制限(64文字)に引っかからないように略称を使う
snapshot_prefix=${snapshot/rds:/snapshot-}
db_cluster_in=${snapshot_prefix}-c-i
db_instance_in=${snapshot_prefix}-i-i
db_cluster_out=${snapshot_prefix}-c-o
db_instance_out=${snapshot_prefix}-i-o

# INPUT DBを作成
aws rds restore-db-cluster-from-snapshot \
    --db-cluster-identifier $db_cluster_in \
    --snapshot-identifier $snapshot  \
    --engine aurora-postgresql \
    --port 5432 \
    --engine-version 16.6 \
    --db-cluster-parameter-group-name $db_cluster_parameter_group_name  \
    --db-subnet-group-name $db_subnet_group_name \
    --vpc-security-group-ids $snapshot_db_security_group \
    --no-copy-tags-to-snapshot
aws rds create-db-instance \
    --db-cluster-identifier $db_cluster_in \
    --db-instance-identifier $db_instance_in  \
    --db-instance-class db.t4g.medium \
    --engine aurora-postgresql \
    --no-auto-minor-version-upgrade

# OUTPUT DBを作成
aws rds restore-db-cluster-from-snapshot \
    --db-cluster-identifier $db_cluster_out \
    --snapshot-identifier $snapshot  \
    --engine aurora-postgresql \
    --port 5432 \
    --engine-version 16.6 \
    --db-cluster-parameter-group-name $db_cluster_parameter_group_name  \
    --db-subnet-group-name $db_subnet_group_name \
    --vpc-security-group-ids $snapshot_db_security_group \
    --no-copy-tags-to-snapshot
aws rds create-db-instance \
    --db-cluster-identifier $db_cluster_out \
    --db-instance-identifier $db_instance_out  \
    --db-instance-class db.t4g.medium \
    --engine aurora-postgresql \
    --no-auto-minor-version-upgrade

# RDSリソース作成後にエラーになった場合のクリーンアップを設定
trap cleanup ERR

echo "[debug] waiting for DB instances to be available"

# rdsが有効になるまでのリトライ回数と待機時間の設定
AVAILABLE_RETRY=120  # 120回 × 60秒 = 120分
AVAILABLE_WAIT=60   # 60秒間隔

echo "[debug] checking INPUT DB instance status..."
for i in $(seq 1 ${AVAILABLE_RETRY}); do
    RESULT=$(aws rds describe-db-instances --db-instance-identifier $db_instance_in --query "DBInstances[0].DBInstanceStatus" --output text 2>/dev/null || echo "unknown")
    echo "[debug] INPUT DB instance status: $RESULT (attempt $i/${AVAILABLE_RETRY})"
    if [ "${RESULT}" = "available" ]; then
        echo "[debug] INPUT DB instance is available"
        break
    fi
    if [ $i -eq ${AVAILABLE_RETRY} ]; then
        echo "[error] INPUT DB instance timeout after ${AVAILABLE_RETRY} attempts"
        exit 1
    fi
    sleep $AVAILABLE_WAIT
done

echo "[debug] checking OUTPUT DB instance status..."
for i in $(seq 1 ${AVAILABLE_RETRY}); do
    RESULT=$(aws rds describe-db-instances --db-instance-identifier $db_instance_out --query "DBInstances[0].DBInstanceStatus" --output text 2>/dev/null || echo "unknown")
    echo "[debug] OUTPUT DB instance status: $RESULT (attempt $i/${AVAILABLE_RETRY})"
    if [ "${RESULT}" = "available" ]; then
        echo "[debug] OUTPUT DB instance is available"
        break
    fi
    if [ $i -eq ${AVAILABLE_RETRY} ]; then
        echo "[error] OUTPUT DB instance timeout after ${AVAILABLE_RETRY} attempts"
        exit 1
    fi
    sleep $AVAILABLE_WAIT
done

echo "[debug] snapshot DB has been launched"

# embulk用の接続情報を設定する
export IN_PGHOST=$(aws rds describe-db-clusters --db-cluster-identifier $db_cluster_in --query="DBClusters[].Endpoint" --output text)
export IN_PGUSER=$SNAPSHOT_DB_USER
export IN_PGPASSWORD=$SNAPSHOT_DB_PASSWORD
export IN_PGDATABASE=$SNAPSHOT_DB_DATABASE
export OUT_PGHOST=$(aws rds describe-db-clusters --db-cluster-identifier $db_cluster_out --query="DBClusters[].Endpoint" --output text)
export OUT_PGUSER=$SNAPSHOT_DB_USER
export OUT_PGPASSWORD=$SNAPSHOT_DB_PASSWORD
export OUT_PGDATABASE=$SNAPSHOT_DB_DATABASE

# psql用の接続情報を設定する
export PGHOST=$OUT_PGHOST
export PGPORT=5432
export PGUSER=$OUT_PGUSER
export PGDATABASE=$OUT_PGDATABASE
export PGPASSWORD=$OUT_PGPASSWORD

echo "[debug] drop unnecessary tables"
psql -c "DROP EXTENSION IF EXISTS pg_stat_statements;"

echo "[debug] truncate tables with unsupported columns"

truncate_tables=(
    "<your_table_names>"
)

psql -c "TRUNCATE TABLE $(printf '%s,' "${truncate_tables[@]}" | sed 's/,$//');"

# embulkを実行する
/home/bin/exec_embulk.sh

echo "[debug] change db name, user name, password"

# マスクが完了したらDB名・ユーザー名・パスワードを本番のものから変更する
old_database=$SNAPSHOT_DB_DATABASE
old_user=$SNAPSHOT_DB_USER
old_password=$SNAPSHOT_DB_PASSWORD
new_database=rentio-development
new_user=default
new_password=password
tmp_database=rentio-tmp
tmp_user=postgres
tmp_password=password

# 作業用の一時的なDBを作る
psql -c "CREATE DATABASE \""$tmp_database\"";"

# 作業用の一時的なユーザーを作る
psql -c "CREATE ROLE $tmp_user WITH PASSWORD '$tmp_password' CREATEDB CREATEROLE LOGIN;"
psql -c "GRANT rds_superuser TO $tmp_user;"

# ユーザー名とパスワードを変更する
export PGUSER=$tmp_user
export PGPASSWORD=$tmp_password
psql -c "ALTER ROLE \""$old_user\"" WITH PASSWORD '$new_password';"
psql -c "ALTER USER \""$old_user\"" RENAME TO \""$new_user\"";"
export PGUSER=$new_user
export PGPASSWORD=$new_password

# DB名を変更する
export PGDATABASE=$tmp_database
psql -c "ALTER DATABASE \""$old_database\"" RENAME TO \""$new_database\"";"
export PGDATABASE=$new_database

# 一時的なユーザーを削除する
psql -c "DROP USER $tmp_user;"

# 一時的なDBを削除する
psql -c "DROP DATABASE \""$tmp_database\"";"

echo "[debug] create snapshot of masked DB"

# マスクしたDBのスナップショットを作成する
aws rds create-db-cluster-snapshot \
    --db-cluster-identifier $db_cluster_out \
    --db-cluster-snapshot-identifier $db_cluster_out \
    --tags "Key=Purpose,Value=seedgen"

echo "[debug] pg_dump seedgen.sql.gz"

# S3にダンプする
pg_dump --no-owner | gzip > /tmp/seedgen.sql.gz
aws s3 cp /tmp/seedgen.sql.gz "s3://<your_bucket_name>/$COPILOT_ENVIRONMENT_NAME-$COPILOT_APPLICATION_NAME-$COPILOT_SERVICE_NAME.sql.gz"

echo "[debug] pg_dump seedgen.sql.gz has been created"

# スナップショットが利用可能になるまで待つ
echo "[debug] waiting for snapshot to be available..."

for i in $(seq 1 ${AVAILABLE_RETRY}); do
    RESULT=$(aws rds describe-db-cluster-snapshots --db-cluster-snapshot-identifier $db_cluster_out --query "DBClusterSnapshots[0].Status" --output text 2>/dev/null || echo "unknown")
    echo "[debug] snapshot status: $RESULT (attempt $i/${AVAILABLE_RETRY})"
    if [ "${RESULT}" = "available" ]; then
        echo "[debug] snapshot is available"
        break
    fi
    if [ $i -eq ${AVAILABLE_RETRY} ]; then
        echo "[error] snapshot timeout after ${AVAILABLE_RETRY} attempts"
        exit 1
    fi
    sleep $AVAILABLE_WAIT
done

echo "[debug] snapshot of masked DB has been created"

echo "[debug] delete snapshot DB"

# DBを削除する
aws rds delete-db-instance \
    --db-instance-identifier $db_instance_in \
    --skip-final-snapshot \
    --delete-automated-backups
aws rds delete-db-instance \
    --db-instance-identifier $db_instance_out \
    --skip-final-snapshot \
    --delete-automated-backups
aws rds delete-db-cluster \
    --db-cluster-identifier $db_cluster_in \
    --skip-final-snapshot \
    --delete-automated-backups
aws rds delete-db-cluster \
    --db-cluster-identifier $db_cluster_out \
    --skip-final-snapshot \
    --delete-automated-backups

こちらはEmbulkの実行スクリプトになります。
liquid形式で書かれたファイルを順番に実行しているだけですね。

bin/exec_embulk.sh

#!/bin/bash
set -e

echo "[debug] exec embulk"

for file in $(find embulk_config -maxdepth 1 -type f -name '*.yml.liquid'); do
    echo "[debug] file: $file"
    java -jar embulk.jar run $file -b embulk_bundle
done

echo "[debug] embulk has been executed successfully"

こちらはローカル実行用のスクリプトになります。
スナップショットを使う必要がないのでかなりシンプルです。
特別にやっていることとしたら、外部キーで参照されているテーブルを編集できるように、トリガを一時的に無効化するくらいですかね。

bin/copy_and_mask_local_db.sh

#!/bin/bash
# psql用の接続情報を設定する

set -e
export PGHOST=$IN_PGHOST
export PGPORT=5432
export PGUSER=$IN_PGUSER
export PGDATABASE=$IN_PGDATABASE
export PGPASSWORD=$IN_PGPASSWORD
psql -c "DROP DATABASE IF EXISTS "\""$OUT_PGDATABASE"\"";"
psql -c "CREATE DATABASE "\""$OUT_PGDATABASE"\"" TEMPLATE "\""$IN_PGDATABASE"\"";"

# 外部キーで参照されているテーブルを編集できるように、トリガーを一時的に無効化する
export PGHOST=$OUT_PGHOST
export PGPORT=5432
export PGUSER=$OUT_PGUSER
export PGDATABASE=$OUT_PGDATABASE
export PGPASSWORD=$OUT_PGPASSWORD
psql -c "select conrelid::regclass::varchar table_name from pg_catalog.pg_constraint r where r.contype = 'f';" -t -A > /tmp/table_names
psql -c "select confrelid::regclass::varchar table_name from pg_catalog.pg_constraint r where r.contype = 'f';" -t -A >> /tmp/table_names
sort /tmp/table_names | uniq | awk '{print "ALTER TABLE "$1" DISABLE TRIGGER ALL;"}' | psql -e

truncate_tables=(
    "<your_table_names>"
)

psql -c "TRUNCATE TABLE $(printf '%s,' "${truncate_tables[@]}" | sed 's/,$//');"

# emublkを実行する
/home/bin/exec_embulk.sh || {
    echo "[error] Embulk failed."
    exit $?
}

# トリガーを有効化して元に戻す
sort /tmp/table_names | uniq | awk '{print "ALTER TABLE "$1" ENABLE TRIGGER ALL;"}' | psql -e

Embulk

Embulkについては、それぞれのテーブルの中身については説明できないので、マスク処理に関するプラグインだけ簡単に説明しようと思います。

.
├── .embulk/
│   └── embulk.properties
├── embulk_bundle/
│   ├── .bundle/
│   │   └── config
│   ├── embulk/
│   │   ├── filter/
│   │   │   └── example.rb
│   │   ├── input/
│   │   │   └── example.rb
│   │   └── output/
│   │       └── example.rb
│   ├── Gemfile
│   └── Gemfile.lock
├── embulk_config/
│   ├── include/
│   │   ├── _in.yml.liquid
│   │   └── _out.yml.liquid
│   └── ...<your_tables>.liquid
└── embulk_plugins/
    └── embulk-filter-anonymize/
        ├── lib/
        │   └── embulk/
        │       └── filter/
        │           └── anonymize.rb
        ├── .ruby-version
        ├── Gemfile
        ├── Rakefile
        └── embulk-filter-anonymize.gemspec

マスク用のブラグインについて

レンティオではマスク用のプラグインとして embulk-filter-anonymize というものを自前で作っています。
マスクの要件を細かく制御したいので独自実装にしており、中身は以下のようになっています。

embulk_plugins/embulk-filter-anonymize/lib/embulk/filter/anonymize.rb

require 'faker'
require 'digest'
require 'digest/md5'

module Embulk
  module Filter

    class AnonymizeFilterPlugin < FilterPlugin
      # filter plugin file name must be: embulk/filter/<name>.rb
      Plugin.register_filter('anonymize', self)

      def self.transaction(config, in_schema, &control)
        task = {
          columns: config.param('columns', :array, default: []),
        }

        out_columns = in_schema

        yield(task, out_columns)
      end

      def initialize(task, in_schema, out_schema, page_builder)
        super

        @columns = {}
        task[:columns].each do |column|
          schema = in_schema.find { |v| v.name == column['name'] }

          next if schema.nil?

          @columns[schema.index] = column
        end
      end

      def close
      end

      def add(page)
        page.each do |record|
          @columns.each do |k, v|
            filtered =
              if v.has_key?('force_value')
                v['force_value']
              elsif v.has_key?('force_nil')
                nil
              elsif v.has_key?('type')
                if record[k].to_s.empty?
                  record[k]
                else
                  fake =
                    case v['type']
                    when 'email'
                      Digest::MD5.hexdigest(record[k].to_s) + '@' + (v.has_key?('domain') ? v['domain'] : 'example.com')
                    when 'md5'
                      Digest::MD5.hexdigest(record[k].to_s)
                    when 'sha256'
                      Digest::SHA256.hexdigest(record[k].to_s)
                    when 'ip'
                      Faker::Internet.ip_v4_address
                    when 'first_name'
                      Faker::Config.locale = v['locale'] if v.has_key?('locale')
                      Faker::Name.first_name
                    when 'last_name'
                      Faker::Config.locale = v['locale'] if v.has_key?('locale')
                      Faker::Name.last_name
                    when 'zip_code'
                      Faker::Config.locale = v['locale'] if v.has_key?('locale')
                      Faker::Address.zip_code
                    when 'street_address'
                      Faker::Config.locale = v['locale'] if v.has_key?('locale')
                      Faker::Address.street_address
                    when 'secondary_address'
                      Faker::Config.locale = v['locale'] if v.has_key?('locale')
                      Faker::Address.secondary_address
                    when 'city'
                      Faker::Config.locale = v['locale'] if v.has_key?('locale')
                      Faker::Address.city
                    when 'company_name'
                      Faker::Config.locale = v['locale'] if v.has_key?('locale')
                      Faker::Company.name
                    else
                      raise ArgumentError
                    end

                  fake = "#{v['prefix']}#{fake}" if v.has_key?('prefix')
                  fake = "#{fake}#{v['suffix']}" if v.has_key?('suffix')
                  fake
                end
              end

            record[k] = filtered
          end

          @page_builder.add(record)
        end
      end

      def finish
        @page_builder.finish
      end
    end

  end
end

まとめ

以上の構成になります。
作成したジョブは定期的に実行されるので、マスクされたDBのスナップショットとS3へのダンプファイルが定期的に作られます。
これを使ってRDSインスタンスを作成すれば、好きな時にDBを立ち上げられます。
また、S3にもpg_dumpされたデータがあるので、こちらをローカルにダウンロードして bin/rails db -p などで実行し、db:migrate すればローカル環境でもマスクされたデータを再現できます。

おまけ: 立ち上げぱなしになっているRDSインスタンスや、スナップショット定期的に削除するLambda function

スナップショットを定期的に作ったり、個人が自由にDBインスタンスを立ち上げられるようになったのは良いですが、消し忘れてお金が無駄にかかってしまうのは避けたいですよね。
そこで、レンティオでは以下のようなLambda Functionを定期的に実行しています。
スナップショットとRDSインスタンスに Purpose: seedgen というタグを設定しているので、その値を元にリソースを定期的に取得して削除するようにしています。

lambda/seedgen-delete-unnecessary-resources/index.py

import boto3

rds = boto3.client('rds')

def main(event, context):
    deleted_resources = []

    # 個人用DB(インスタンス)を削除
    response = rds.describe_db_instances()
    for instance in response['DBInstances']:
        db_instance_identifier = instance['DBInstanceIdentifier']
        if db_instance_identifier.startswith('seedgen-') and {'Key': 'Purpose', 'Value': 'seedgen'} in instance['TagList']:
            print(f"Deleting {db_instance_identifier}")
            rds.delete_db_instance(
                DBInstanceIdentifier=db_instance_identifier,
                SkipFinalSnapshot=True,
            )
            deleted_resources.append(db_instance_identifier)

    # 個人用DB(クラスタ)を削除
    response = rds.describe_db_clusters()
    for cluster in response['DBClusters']:
        db_cluster_itentifier = cluster['DBClusterIdentifier']
        if db_cluster_itentifier.startswith('seedgen-') and {'Key': 'Purpose', 'Value': 'seedgen'} in cluster['TagList']:
            print(f"Deleting {db_cluster_itentifier}")
            rds.delete_db_cluster(
                DBClusterIdentifier=db_cluster_itentifier,
                SkipFinalSnapshot=True,
            )
            deleted_resources.append(db_cluster_itentifier)

    # 新しいスナップショット3つ残して、それより古いものは削除
    snapshot_identifiers = list(map(
        lambda v: v['DBClusterIdentifier'],
        sorted(
            filter(
                lambda v: v['DBClusterIdentifier'].startswith('snapshot-dbcluster-rds-') and {'Key': 'Purpose', 'Value': 'seedgen'} in v['TagList'],
                rds.describe_db_cluster_snapshots(SnapshotType='manual')['DBClusterSnapshots']
            ),
            key=lambda v: v['SnapshotCreateTime'],
            reverse=True
        )
    ))[3:]
    for snapshot_identifier in snapshot_identifiers:
        print(f"Deleting {snapshot_identifier}")
        rds.delete_db_cluster_snapshot(DBClusterSnapshotIdentifier=snapshot_identifier)
        deleted_resources.append(snapshot_identifier)

    if len(deleted_resources) == 0:
        print("No resources to delete")

    return { 'deleted_resources': deleted_resources }

採用情報

レンティオでは絶賛、エンジニアを募集しています！
https://www.wantedly.com/companies/rentio