翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

チュートリアル: VPC 内で ElastiCache にアクセスするための Lambda の設定

このチュートリアルでは、ElastiCache サーバーレスキャッシュの作成、Lambda 関数の作成、Lambda 関数のテスト、オプションのクリーンアップ方法について説明します。

ステップ 1: ElastiCache サーバーレスキャッシュを作成する。

新しいサーバーレスキャッシュを作成するには、次の手順に従います。

ステップ 1.1: サーバーレスキャッシュを作成する

このステップでは、 AWS Command Line Interface (CLI) を使用して、アカウントの us-east-1 リージョンのデフォルトの Amazon VPC にサーバーレスキャッシュを作成します。ElastiCache コンソールまたは API を使用してサーバーレスキャッシュを作成する方法については、「Redis OSS サーバーレスキャッシュを作成する」を参照してください。

aws elasticache create-serverless-cache \
  --serverless-cache-name cache-01  \
--description "ElastiCache IAM auth application" \
--engine valkey

[ステータス] フィールドの値が CREATING に設定されていることに注意してください。ElastiCache がキャッシュの作成を完了するまで、1 分かかる場合があります。

ステップ 1.2: サーバーレスキャッシュエンドポイントをコピーする

describe-serverless-caches コマンドを使用して、ElastiCache for Redis OSS がキャッシュの作成を完了したことを確認します。

aws elasticache describe-serverless-caches \
--serverless-cache-name cache-01 

出力に表示されたエンドポイントアドレスをコピーします。Lambda 関数のデプロイパッケージを作成するときに、このアドレスが必要になります。

ステップ 1.3: IAM ロールを作成する

ステップ 1.4: デフォルトユーザーを作成する

ステップ 2: ElastiCache 用の Lambda 関数を作成する

ElastiCache キャッシュにアクセスするための Lambda 関数を作成するには、以下の手順に従います。

ステップ 2.1: Lambda 関数を作成する

このチュートリアルでは、Lambda 関数のコード例を Python で示します。

Python

次の Python のコード例では、ElastiCache キャッシュに対して項目の読み取り/書き込みを行います。コードを app.py という名前のファイルに保存します。コード内の elasticache_endpoint 値を、前のステップでコピーしたエンドポイントアドレスに必ず置き換えてください。

from typing import Tuple, Union
from urllib.parse import ParseResult, urlencode, urlunparse

import botocore.session
import redis
from botocore.model import ServiceId
from botocore.signers import RequestSigner
from cachetools import TTLCache, cached
import uuid

class ElastiCacheIAMProvider(redis.CredentialProvider):
    def __init__(self, user, cache_name, is_serverless=False, region="us-east-1"):
        self.user = user
        self.cache_name = cache_name
        self.is_serverless = is_serverless
        self.region = region

        session = botocore.session.get_session()
        self.request_signer = RequestSigner(
            ServiceId("elasticache"),
            self.region,
            "elasticache",
            "v4",
            session.get_credentials(),
            session.get_component("event_emitter"),
        )

    # Generated IAM tokens are valid for 15 minutes
    @cached(cache=TTLCache(maxsize=128, ttl=900))
    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
        query_params = {"Action": "connect", "User": self.user}
        if self.is_serverless:
            query_params["ResourceType"] = "ServerlessCache"
        url = urlunparse(
            ParseResult(
                scheme="https",
                netloc=self.cache_name,
                path="/",
                query=urlencode(query_params),
                params="",
                fragment="",
            )
        )
        signed_url = self.request_signer.generate_presigned_url(
            {"method": "GET", "url": url, "body": {}, "headers": {}, "context": {}},
            operation_name="connect",
            expires_in=900,
            region_name=self.region,
        )
        # RequestSigner only seems to work if the URL has a protocol, but
        # Elasticache only accepts the URL without a protocol
        # So strip it off the signed URL before returning
        return (self.user, signed_url.removeprefix("https://"))

def lambda_handler(event, context):
    username = "iam-user-01" # replace with your user id
    cache_name = "cache-01" # replace with your cache name
    elasticache_endpoint = "cache-01-xxxxx.serverless.use1.cache.amazonaws.com" # replace with your cache endpoint
    creds_provider = ElastiCacheIAMProvider(user=username, cache_name=cache_name, is_serverless=True)
    redis_client = redis.Redis(host=elasticache_endpoint, port=6379, credential_provider=creds_provider, ssl=True, ssl_cert_reqs="none")
    
    key='uuid'
    # create a random UUID - this will be the sample element we add to the cache
    uuid_in = uuid.uuid4().hex
    redis_client.set(key, uuid_in)
    result = redis_client.get(key)
    decoded_result = result.decode("utf-8")
    # check the retrieved item matches the item added to the cache and print
    # the results
    if decoded_result == uuid_in:
        print(f"Success: Inserted {uuid_in}. Fetched {decoded_result} from Valkey.")
    else:
        raise Exception(f"Bad value retrieved. Expected {uuid_in}, got {decoded_result}")
        
    return "Fetched value from Valkey"

このコードでは、Python redis-py ライブラリを使用して項目をキャッシュに格納し、取得します。このコードでは、cachetools を使用して、生成された IAM 認証トークンを 15 分間キャッシュします。redis-py と cachetools を含むデプロイパッケージを作成するには、次のステップを実行します。

app.py ソースコードファイルを含むプロジェクトディレクトリに、redis-py と cachetools ライブラリをインストールするフォルダパッケージを作成します。

mkdir package

pip を使用して redis-py、cachetools をインストールします。

pip install --target ./package redis
pip install --target ./package cachetools

redis-py ライブラリと cachetools ライブラリを含む .zip ファイルを作成します。Linux および macOS では、次のコマンドを実行します。Windows では、任意の zip ユーティリティを使用して、redis-py ライブラリと cachetools ライブラリをルートに置く .zip ファイルを作成します。

cd package
zip -r ../my_deployment_package.zip .

.zip ファイルに関数コードを追加します。Linux および macOS では、次のコマンドを実行します。Windows では、任意の zip ユーティリティを使用して .zip ファイルのルートに app.py を追加します。

cd ..
zip my_deployment_package.zip app.py

ステップ 2.2: IAM ロール (実行ロール) を作成する

という名前 AWS の管理ポリシーをロールAWSLambdaVPCAccessExecutionRoleにアタッチします。

aws iam attach-role-policy \
 --role-name "elasticache-iam-auth-app" \
 --policy-arn "arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole"

ステップ 2.3: デプロイパッケージをアップロードする (Lambda 関数を作成する)

このステップでは、create-function AWS CLI コマンドを使用して Lambda 関数 (AccessValkey) を作成します。

デプロイパッケージの.zip ファイルを含むプロジェクトディレクトリから、次の Lambda CLI create-function コマンドを実行します。

ロールのオプションには、前のステップで作成した実行ロールの ARN を使用します。vpc-config には、デフォルト VPC のサブネットとデフォルト VPC のセキュリティグループ ID をカンマで区切ったリストを入力します。これらの値は Amazon VPC コンソール にあります。デフォルトの VPC のサブネットを検索するには、VPCs を選択し、 AWS アカウントのデフォルトの VPC を選択します。この VPC のセキュリティグループを確認するには、[セキュリティ] に移動して [セキュリティグループ] を選択します。us-east-1 リージョンが選択されていることを確認します。

aws lambda create-function \
--function-name AccessValkey  \
--region us-east-1 \
--zip-file fileb://my_deployment_package.zip \
--role arn:aws:iam::123456789012:role/elasticache-iam-auth-app \
--handler app.lambda_handler \
--runtime python3.12  \
--timeout 30 \
--vpc-config SubnetIds=comma-separated-vpc-subnet-ids,SecurityGroupIds=default-security-group-id

ステップ 3: ElastiCache で Lambda 関数をテストする

このステップでは、invoke コマンドを使用して Lambda 関数を手動で呼び出します。Lambda 関数を実行すると、UUID が生成され、Lambda コードで指定した ElastiCache キャッシュにその UUID が書き込まれます。次に、Lambda 関数はキャッシュから項目を取得します。

ステップ 4: クリーンアップする (オプション)

クリーンアップするには、以下の手順を実行します。

ステップ 4.1: Lambda 関数を削除する

aws lambda delete-function \
 --function-name AccessValkey 

ステップ 4.2: サーバーレスキャッシュを削除する

キャッシュを削除します。

aws elasticache delete-serverless-cache \
 --serverless-cache-name cache-01 

ユーザーとユーザーグループを削除します。

aws elasticache delete-user \
 --user-id default-user-disabled

aws elasticache delete-user \
 --user-id iam-user-01

aws elasticache delete-user-group \
 --user-group-id iam-user-group-01

ステップ 4.3: IAM ロールとポリシーを削除する

aws iam detach-role-policy \
 --role-name "elasticache-iam-auth-app" \
 --policy-arn "arn:aws:iam::123456789012:policy/elasticache-allow-all"
 
aws iam detach-role-policy \
--role-name "elasticache-iam-auth-app" \
--policy-arn "arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole"
 
aws iam delete-role \
 --role-name "elasticache-iam-auth-app"
  
 aws iam delete-policy \
  --policy-arn "arn:aws:iam::123456789012:policy/elasticache-allow-all"