プロジェクトでAWS CDKで開発を行う機会があったため、開発で得た知識を発信しようと思います。
今回の記事では、CDK(ver.2) Pythonを使って、同じプライベートサブネット内にあるAWS Lambda関数からRDS Proxyを介してプライベートサブネット内にあるRDSデータベースに接続する方法を解説します。

全コードはGithubを参照してください
Githubはここ

前提条件

AWS CDK v.2 がインストールされていること
Python 3.x がインストールされていること
AWSアカウントがあり、AWS CLIが設定されていること

※Cloud9を使うとこの辺りがPassできるため、Cloud9を使って今回の記事の内容は作成しています。

構築手順

1. CDKアプリの初期化

先ずはCDKアプリの初期化を行います。

$ mkdir cdk-lambda-rds-proxy
$ cd cdk-lambda-rds-proxy
$ cdk init --language python

2. 必要なパッケージをインストール

ここでは、CDKアプリを初期化した際に作成された.venvを有効化しています。

$ source .venv/bin/activate
$ pip install -r requirements.txt

3. スタックの作成

cdk_lambda_rds_proxy_stack.py

from constructs import Construct
from aws_cdk import (
    Stack,
    RemovalPolicy,
    aws_ec2 as ec2,
    aws_lambda as _lambda,
    aws_rds as rds,
    aws_secretsmanager as sm,
)


class CdkLambdaRdsProxyStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # VPCの作成
        vpc = ec2.Vpc(self, "MyVPC",
            max_azs=2
        )

        # RDSのセキュリティグループの作成
        rds_security_group = ec2.SecurityGroup(
            self, 
            "RdsSecurityGroup",
            vpc=vpc,
            description="Allow Lambda access to RDS"
        )

        # RDSの作成
        rds_instance = rds.DatabaseInstance(
            self, 
            "RDS",
            engine=rds.DatabaseInstanceEngine.postgres(
                version=rds.PostgresEngineVersion.VER_14_5
            ),
            vpc=vpc,
            security_groups=[rds_security_group],
            instance_type=ec2.InstanceType.of(
                ec2.InstanceClass.BURSTABLE3, ec2.InstanceSize.SMALL
            ),
            allocated_storage=20,
            removal_policy=RemovalPolicy.DESTROY,
            deletion_protection=False,
            database_name="MyDatabase",
            auto_minor_version_upgrade=True,
            multi_az=False
        )

        # Lambdaのセキュリティグループの作成
        lambda_security_group = ec2.SecurityGroup(
            self, 
            "LambdaSecurityGroup",
            vpc=vpc,
            description="Allow Lambda to access RDS Proxy"
        )

        # RDS Proxyの作成
        rds_proxy = rds.DatabaseProxy(
            self, 
            "RDSProxy",
            proxy_target=rds.ProxyTarget.from_instance(rds_instance),
            vpc=vpc,
            security_groups=[lambda_security_group],
            db_proxy_name="MyRDSProxy",
            debug_logging=False,
            secrets=[rds_instance.secret],
            require_tls=True
        )
        
        # psycopg2のLambda Layer
        psycopg2_layer = _lambda.LayerVersion.from_layer_version_arn(
            self, 
            "Psycoog2Layer",
            'arn:aws:lambda:ap-northeast-1:898466741470:layer:psycopg2-py38:1'
        )
        
        # Lambda関数の作成
        lambda_function = _lambda.Function(
            self, 
            "LambdaFunction",
            runtime=_lambda.Runtime.PYTHON_3_8,
            handler="lambda_function.handler",
            code=_lambda.Code.from_asset("cdk_lambda_rds_proxy/lambda"),
            layers=[psycopg2_layer],
            vpc=vpc,
            security_groups=[lambda_security_group],
            vpc_subnets=ec2.SubnetSelection(subnets=vpc.private_subnets),
            environment={
                "DB_SECRET_ARN": rds_instance.secret.secret_arn,
                "DB_PROXY_ENDPOINT": rds_proxy.endpoint
            }
        )
    
        # RDSインスタンスとLambda関数のアクセス許可
        rds_instance.connections.allow_from(
            lambda_security_group,
            port_range=ec2.Port.tcp(5432)
        )
        rds_proxy.connections.allow_from(
            lambda_security_group,
            port_range=ec2.Port.tcp(5432)
        )
        
        # Secretへのアクセス許可
        rds_instance.secret.grant_read(lambda_function)

4. Lambda関数の作成

lambdaフォルダを作成し、その中にlambda_function.pyファイルを作成して、以下の内容を記述する。

lambda_function.py

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

def get_db_credentials(secret_arn):
    secrets_manager = boto3.client("secretsmanager")
    try:
        response = secrets_manager.get_secret_value(SecretId=secret_arn)
        return json.loads(response["SecretString"])
    except ClientError as e:
        print(e)
        return None

def handler(event, context):
    secret_arn = os.environ["DB_SECRET_ARN"]
    db_proxy_endpoint = os.environ["DB_PROXY_ENDPOINT"]

    credentials = get_db_credentials(secret_arn)

    if not credentials:
        return {
            "statusCode": 500,
            "body": "Error retrieving database credentials"
        }
    
    connection_string = f"dbname={credentials['dbname']} user={credentials['username']} password={credentials['password']} host={db_proxy_endpoint} port={credentials['port']} sslmode=require"
    
    try:
        conn = psycopg2.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute("SELECT version();")
        result = cursor.fetchone()

        return {
            "statusCode": 200,
            "body": f"Database version: {result[0]}"
        }

    except Exception as e:
        print(e)
        return {
            "statusCode": 500,
            "body": "Error connecting to the database"
        }

5. CDKアプリをデプロイ

Stackの準備が完了したため、デプロイします。

$ cdk deploy

デプロイが完了すると、Lambda関数からRDSデータベースに接続してバージョン情報を取得できるようになります。

↓↓ 実際に接続確認した結果

まとめ

この記事では、CDK(ver.2) Pythonを使って、同じプライベートサブネット内にあるLambda関数からRDS Proxyを介してプライベートサブネット内にあるRDSデータベースに接続する方法を記載しました。
LambdaとRDSを直接つなぐことはアンチパターンとなるようですので、Lambda＋RDSの構成を取る際は参考にしてみてください。

参考

AWS CDK API Reference