はじめに

本日は RDS Proxy を使ってみようと思います。RDS Proxyは AWS認定試験で頻出するサービスなのでしっかり押さえておきたいですね。

RDS Proxy とは

Lambda 関数から DB に多くの接続が発生する可能性があるアーキテクチャの場合、リクエストの増加に伴い DB 接続の拒否が発生しますが、RDS Proxy を使うことによって DB 接続プールを確立し再利用することで効率的な接続をすることができるようになります。

RDS Proxy を使わないパターン

RDS Proxy を使うパターン

アーキテクチャ

Amazon Secrets Manager に格納した DB 接続情報を使って Lambda から Aurora に接続します。

手順

DB の作成
IAMロールの作成
Secrets Manager の設定
RDS Proxy の作成
テストデータ作成
Lambda 関数の作成
動作確認

1. DB の作成

以下の設定で DB を作成します。

[データベース作成方法を選択]：標準作成
[エンジンのタイプ]：Amazon Aurora [エディション]：PostgreSQL との互換性を持つ Amazon Aurora
[テンプレート]：開発/テスト
[DB クラスター識別子]：aws-room-cluster
[マスターユーザー名]：postgres
[パスワードの自動生成]：チェックしない
[DB インスタンスクラス]：メモリ最適化クラス
[可用性と耐久性]：Aurora レプリカを作成しない
[ネットワークタイプ]：IPv4
[Virtual Private Cloud (VPC)]：任意のVPC
[サブネットグループ]：任意のサブネットグループ
[パブリックアクセス]：あり
[VPCセキュリティグループ]：既存の選択
[既存のセキュリティグループ ]：任意のセキュリティグループ
[アベイラビリティーゾーン]：任意のアベイラビリティーゾーン

DBを作成したら接続確認をします。

2. IAMロールの作成

「access-rdb」という Amazon Aurora と AWS Secrets Manager への権限を付与したロールを作成します。

3. Secrets Manager の設定

以下のようにシークレットを作成します。

[シークレットのタイプ]：Amazon RDS データベースの認証情報
[ユーザー名]：「1. DB の作成」で設定したユーザ
[パスワード]：「1. DB の作成」で設定したパスワード
[暗号化キー]：任意の暗号化キー
[データベース]：「1. DB の作成」で作成したデータベース

4. RDS Proxy の作成

以下のように RDS Proxy を作成します。

[エンジンファミリー]：PostgreSQL
[プロキシ識別子]：任意の文字列
[データベース]：「1. DB の作成」で作成したAurora DB クラスター
[接続プールの最大接続数]：100
[Secrets Manager のシークレット]：「4. Secrets Manager の設定」で作成したシークレット
[IAM ロール]：IAMロールを作成
[IAM 認証]：許可されていません

5. テストデータ作成

test というテーブルに以下のようなデータを作成しました。

6. Lambda 関数の作成

以下のように Lambda 関数を作成します。

[オプション]：一から作成
[関数名]：access-rds-through-proxy
[ランタイム]：Python 3.8
[アーキテクチャ]：x86_64
[既存のロール]：access-rdb

レイヤーの追加

psycopg2 を使用して PostgreSQL に接続するために以下のようにレイヤーの追加を行います。

[レイヤーソース]：ARN を指定
[ARNを指定]：arn:aws:lambda:ap-northeast-1:898466741470:layer:psycopg2-py38:1

ソースコード

Secrets Manager からDB 接続情報を取得して作成したデータベースにあるテーブルにアクセスします。

import boto3
from botocore.exceptions import ClientError
import base64
import psycopg2
import json

secret_name = "aurora"
region_name = "ap-northeast-1"
    
def lambda_handler(event, context):
    secret_dict = json.loads(get_secret())

    connect = psycopg2.connect(
                  host = secret_dict['host'],
                  user = secret_dict['username'],
                  password = secret_dict['password'],
                  port = 5432,
                  dbname = 'postgres'
              )
    with connect.cursor() as cur:
        cur.execute('select * from public.test')
        rows = cur.fetchall()

    print(rows)

def get_secret():
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )

    except ClientError as e:
        print(e.response['Error']['Code'])
        if e.response['Error']['Code'] == 'DecryptionFailureException':
            raise e
        elif e.response['Error']['Code'] == 'InternalServiceErrorException':
            raise e
        elif e.response['Error']['Code'] == 'InvalidParameterException':
            raise e
        elif e.response['Error']['Code'] == 'InvalidRequestException':
            raise e
        elif e.response['Error']['Code'] == 'ResourceNotFoundException':
            raise e
    else:
        if 'SecretString' in get_secret_value_response:
            secret = get_secret_value_response['SecretString']
        else:
            secret = base64.b64decode(get_secret_value_response['SecretBinary'])

        return secret