Lambda AZメタデータで同一AZルーティングを実現する — ElastiCacheのレイテンシを半減

curl -s \
  -H "Authorization: Bearer ${AWS_LAMBDA_METADATA_TOKEN}" \
  "http://${AWS_LAMBDA_METADATA_API}/2026-01-15/metadata/execution-environment"

{
  "AvailabilityZoneID": "apne1-az2"
}

{
  "az_id": "apne1-az2",
  "latency_ms": 180.221,
  "headers": {
    "Cache-Control": "private, max-age=43200, immutable",
    "Content-Type": "application/json",
    "Content-Length": "34"
  }
}

az-metadata-test-001 | ap-northeast-1a (apne1-az4) | primary
az-metadata-test-002 | ap-northeast-1c (apne1-az1) | replica
az-metadata-test-003 | ap-northeast-1d (apne1-az2) | replica

import json
import os
import socket
import ssl
import time
import urllib.request
 
 
def get_az_id():
    api = os.environ["AWS_LAMBDA_METADATA_API"]
    token = os.environ["AWS_LAMBDA_METADATA_TOKEN"]
    url = f"http://{api}/2026-01-15/metadata/execution-environment"
    req = urllib.request.Request(
        url, headers={"Authorization": f"Bearer {token}"}
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())["AvailabilityZoneID"]
 
 
def measure_valkey_latency(host, port, iterations=50):
    ctx = ssl.create_default_context()
    sock = socket.create_connection((host, port), timeout=5)
    tls_sock = ctx.wrap_socket(sock, server_hostname=host)
    latencies = []
    try:
        for _ in range(iterations):
            start = time.perf_counter()
            tls_sock.sendall(b"*1\r\n$4\r\nPING\r\n")
            tls_sock.recv(64)
            latencies.append((time.perf_counter() - start) * 1000)
    finally:
        tls_sock.close()
    latencies.sort()
    return {
        "avg_ms": round(sum(latencies) / len(latencies), 3),
        "p50_ms": round(latencies[len(latencies) // 2], 3),
        "min_ms": round(latencies[0], 3),
    }
 
 
def handler(event, context):
    lambda_az_id = get_az_id()
 
    # 環境変数からノード情報を取得
    # 形式: cluster_id|address|port|az_id|az_name|role,...
    nodes = []
    for entry in os.environ["CACHE_NODES"].split(","):
        parts = entry.split("|")
        nodes.append({
            "id": parts[0], "address": parts[1], "port": int(parts[2]),
            "az_id": parts[3], "az_name": parts[4], "role": parts[5],
        })
 
    results = {"lambda_az_id": lambda_az_id, "same_az": [], "cross_az": []}
    for node in nodes:
        latency = measure_valkey_latency(node["address"], node["port"])
        entry = {**node, "latency": latency}
        if node["az_id"] == lambda_az_id:
            results["same_az"].append(entry)
        else:
            results["cross_az"].append(entry)
    return results

import json
import os
import urllib.request
 
# Init フェーズで一度だけ取得してキャッシュ
_az_id = None
 
def get_lambda_az_id():
    global _az_id
    if _az_id is None:
        api = os.environ["AWS_LAMBDA_METADATA_API"]
        token = os.environ["AWS_LAMBDA_METADATA_TOKEN"]
        url = f"http://{api}/2026-01-15/metadata/execution-environment"
        req = urllib.request.Request(
            url, headers={"Authorization": f"Bearer {token}"}
        )
        with urllib.request.urlopen(req) as resp:
            _az_id = json.loads(resp.read())["AvailabilityZoneID"]
    return _az_id
 
def select_same_az_endpoint(endpoints, lambda_az_id):
    """同一AZのエンドポイントを優先し、なければフォールバック。"""
    same_az = [ep for ep in endpoints if ep["az_id"] == lambda_az_id]
    return same_az[0] if same_az else endpoints[0]

from aws_lambda_powertools.utilities.lambda_metadata import get_lambda_metadata
 
def handler(event, context):
    metadata = get_lambda_metadata()
    az_id = metadata.availability_zone_id  # e.g., "apne1-az2"
    # ...

# Lambda関数削除
aws lambda delete-function --function-name az-routing-test
 
# ElastiCacheレプリケーショングループ削除
aws elasticache delete-replication-group \
  --replication-group-id az-metadata-test \
  --no-retain-primary-cluster
# → 完了まで数分待つ
 
# ElastiCacheサブネットグループ削除
aws elasticache delete-cache-subnet-group \
  --cache-subnet-group-name az-test-subnet
 
# セキュリティグループ削除
aws ec2 delete-security-group --group-id <SG_ID>
 
# IAMロール削除
aws iam detach-role-policy --role-name lambda-az-test-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
aws iam delete-role --role-name lambda-az-test-role