部署时智能在 AWS CDK：自定义资源实战

I’m happy to help translate the article, but I need the text you’d like translated. Could you please paste the content (excluding the source line you already provided) here? Once I have the article text, I’ll translate it into Simplified Chinese while preserving the original formatting, markdown, and any code blocks.

介绍

在真实的 AWS 平台中，单个 CDK 代码库通常会部署到多个 AWS 账户，每个账户代表不同的环境，例如 development（开发）、staging（预发布）或 production（生产）。

虽然 AWS CDK 在定义基础设施方面表现出色，但它有一个局限性：

它无法在部署时根据存储在账户内部的值（例如存储在 AWS Systems Manager（SSM）参数存储中的值）做出决策。

在本博客中，我们将使用 Lambda 支持的自定义资源 来解决一个实际的平台工程问题，实现 环境感知的决策，用于安装 EKS Helm 插件。

实际问题

您是一名平台工程师，负责管理跨多个 AWS 账户的 EKS 集群：

贵组织已经在 SSM 参数中集中存储环境类型：

/platform/account/env

其取值如下：

• development
• staging
• production

现在您想在每个 EKS 集群上安装 ingress‑nginx，但需要进行不同的配置：

• development → replicaCount = 1
• staging / production → replicaCount = 2

为什么不使用 CDK Context？（简短版）

乍一看，CDK context 变量似乎是实现基于环境的配置的更简单方案。然而，context 值在 合成阶段 解析，而不是在部署阶段。这意味着它们必须在外部提供（通过 cdk.json 或 CI/CD 流水线），并且无法感知 账户级元数据，例如存储在 SSM Parameter Store 中的值。

在多账户平台中，这通常会导致：

• 手动协调
• 配置漂移
• 治理问题

由于环境分类已经存在于 AWS 账户内部，并且应由平台统一管理，使用 部署时自定义资源 可以确保配置的准确性、一致性以及中心化控制。

为什么仅使用 CDK 不够

AWS CDK 在 合成阶段 评估逻辑，但 SSM 参数值只能在 部署阶段 可靠获取。因此你无法：

• 对该值使用 CDK if 语句
• 在堆栈中硬编码环境值
• 可靠地依赖 CDK 上下文

你需要的是 部署时逻辑.

解决方案：基于 Lambda 的自定义资源

A Custom Resource allows CloudFormation to:

1. 在堆栈创建或更新期间调用 Lambda 函数
2. 等待结果
3. 将返回的属性作为其他资源的输入

In this case, the Custom Resource:

• 从 SSM 读取环境值
• 计算正确的 Helm 值
• 将其返回给 CDK
• CDK 将其传递给 Helm 图表

架构概览

部署流程

1. CDK 创建
   • EKS 集群
   • SSM 参数 /platform/account/env
   • Lambda 函数
2. 自定义资源触发 Lambda
3. Lambda 计算 Helm 值
4. 使用返回的值安装 Helm Chart

这确保：

• 单一 CDK 代码库
• 零手动步骤
• 环境感知的行为

Source:

CDK 堆栈代码（Python）

下面是 CDK 堆栈，它会创建：

• EKS 集群
• SSM 参数
• Lambda 函数
• 自定义资源
• 带动态值的 Helm 图表

from aws_cdk import (
    Stack,
    aws_eks as eks,
    aws_ec2 as ec2,
    aws_iam as iam,
    aws_ssm as ssm,
    aws_signer as signer,
    aws_lambda as _lambda,
    custom_resources as cr,
    CustomResource,
    Token,
)
from aws_cdk.lambda_layer_kubectl_v34 import KubectlV34Layer
from constructs import Construct
import os


class TestStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, *, vpc: ec2.IVpc = None, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        aws_account_id = self.node.try_get_context("aws-account-id")

        # -------------------------------------------------
        # EKS 集群
        # -------------------------------------------------
        cluster = eks.Cluster(
            self,
            "MyEKS",
            version=eks.KubernetesVersion.V1_34,
            endpoint_access=eks.EndpointAccess.PUBLIC_AND_PRIVATE,
            default_capacity=0,
            default_capacity_instance=ec2.InstanceType.of(
                ec2.InstanceClass.T3,
                ec2.InstanceSize.MEDIUM,
            ),
            kubectl_layer=KubectlV34Layer(self, "kubectl"),
            vpc=vpc,
            vpc_subnets=[
                ec2.SubnetSelection(
                    subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
                )
            ],
            cluster_name="MyEKS",
            tags={"Name": "MyEKS", "Purpose": "Swisscom-Interview"},
        )

        # -------------------------------------------------
        # EKS 管理员访问
        # -------------------------------------------------
        admin_user = iam.User(self, "EKSAdmin")
        cluster.aws_auth.add_user_mapping(admin_user, groups=["system:masters"])

        # -------------------------------------------------
        # 将环境存储在 SSM 中
        # -------------------------------------------------
        ssm.StringParameter(
            self,
            "MyEnvParam",
            parameter_name="/platform/account/env",
            string_value="development",
            description="Environment Name",
        )

        # -------------------------------------------------
        # Lambda 代码签名
        # -------------------------------------------------
        signing_profile = signer.SigningProfile(
            self,
            "SigningProfile",
            platform=signer.Platform.AWS_LAMBDA_SHA384_ECDSA,
        )

        code_signing_config = _lambda.CodeSigningConfig(
            self,
            "CodeSigningConfig",
            signing_profiles=[signing_profile],
        )

        # -------------------------------------------------
        # Lambda 函数（读取 SSM 参数）
        # -------------------------------------------------
        fn = _lambda.Function(
            self,
            "MySSMParamLambda",
            runtime=_lambda.Runtime.PYTHON_3_13,
            handler="index.lambda_handler",
            code=_lambda.Code.from_asset(
                os.path.join(os.path.dirname(__file__), "lambda_functions")
            ),
            environment={"SSM_PARAM_NAME": "/platform/account/env"},
            code_signing_config=code_signing_config,
        )

        fn.add_to_role_policy(
            iam.PolicyStatement(
                actions=["ssm:GetParameter"],
                resources=["*"],  # 在生产环境中请调整为特定参数 ARN
            )
        )

        # -------------------------------------------------
        # 调用 Lambda 的自定义资源
        # -------------------------------------------------
        custom_resource = cr.AwsCustomResource(
            self,
            "EnvLookupCustomResource",
            on_create=cr.AwsSdkCall(
                service="Lambda",
                action="invoke"

                parameters={
                    "FunctionName": fn.function_name,
                    "Payload": Token.as_string(
                        {
                            "RequestType": "Create",
                            "ResourceProperties": {"SSM_PARAM_NAME": "/platform/account/env"},
                        }
                    ),
                },
                physical_resource_id=cr.PhysicalResourceId.of("EnvLookup"),
            ),
            policy=cr.AwsCustomResourcePolicy.from_sdk_calls(
                resources=cr.AwsCustomResourcePolicy.ANY_RESOURCE
            ),
        )

        # -------------------------------------------------
        # Use the value returned by the Custom Resource in the Helm chart
        # -------------------------------------------------
        replica_count = custom_resource.get_response_field("Payload.body.replicaCount")

        eks.HelmChart(
            self,
            "IngressNginx",
            cluster=cluster,
            chart="ingress-nginx",
            repository="https://kubernetes.github.io/ingress-nginx",
            release="ingress-nginx",
            namespace="ingress-nginx",
            values={"controller": {"replicaCount": replica_count}},
        )

Lambda 代码（未显示）读取 /platform/account/env，将该值映射到相应的 replicaCount，并在响应负载中返回它。

CDK 代码（基础设施）

resources = [
    f"arn:aws:ssm:{self.region}:{self.account}:parameter/platform/account/env"
]

# Custom Resource Provider
provider = cr.Provider(
    self,
    "EnvToHelmProvider",
    on_event_handler=fn,
)

env_cr = CustomResource(
    self,
    "EnvToHelmValues",
    service_token=provider.service_token,
)

replica_count = Token.as_number(
    env_cr.get_att("ReplicaCount")
)

# Install ingress‑nginx Helm chart
cluster.add_helm_chart(
    "nginx-ingress",
    chart="nginx-ingress",
    repository="https://helm.nginx.com/stable",
    namespace="kube-system",
    values={
        "controller": {
            "replicaCount": replica_count
        }
    },
)

Lambda 代码（自定义资源逻辑）

• 从 SSM 读取环境
• 计算正确的副本数
• 将值返回给 CloudFormation

import boto3
import os
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

ssm = boto3.client("ssm")
PARA_NAME = os.environ["SSM_PARAM_NAME"]


def get_parameter(para_name):
    parameter = ssm.get_parameter(Name=para_name)
    env = parameter["Parameter"]["Value"].strip().lower()

    if env == "development":
        value = 1
    elif env in ["staging", "production"]:
        value = 2
    else:
        raise ValueError(
            f"Invalid environment {env} in SSM Parameter {para_name}"
        )

    logger.info(f"Computed replicaCount={value} for env={env}")

    return {
        "Environment": env,
        "ReplicaCount": value,
    }


def lambda_handler(event, context):
    if event.get("RequestType") == "Delete":
        return {
            "PhysicalResourceId": event.get("PhysicalResourceId", "env"),
            "Data": {},
        }

    data = get_parameter(PARA_NAME)

    return {
        "PhysicalResourceId": f"{PARA_NAME}:{data['Environment']}",
        "Data": data,
    }

为什么这种模式效果好

此方法提供：

• 单一 CDK 代码库
• 环境感知行为
• 无需手动 Helm 覆盖
• 部署时决策
• 可测试的业务逻辑

它在添加更多插件时仍能良好扩展，例如：

• cert‑manager
• external‑dns
• cluster‑autoscaler
• logging agents

结论

AWS CDK 本质上是声明式的，但真实平台需要 部署时智能。通过将 Lambda 支持的自定义资源 与 CDK 结合使用，您可以基于 真实的账户元数据 而不是硬编码的假设来做基础设施决策。

此模式是平台团队实现 一致性、安全性和跨环境自动化 的强大工具。