AWS CDK:创建工具类以避免跨栈引用问题
当然可以,完全不需要把这个逻辑做成CDK Stack。你可以把它封装成独立的Construct类(符合CDK最佳实践)或者纯工具函数,既能从SSM Parameter Store读取配置,又能避免跨引用问题,还能在多个仓库的Stack中复用。
方案一:封装为独立的Construct类(推荐)
这个类继承cdk.Construct而非cdk.Stack,职责是从SSM读取VPC相关参数、生成ec2.IVpc实例供其他Stack使用,本身不会部署任何资源。
from aws_cdk import aws_ec2 as ec2 from aws_cdk import aws_ssm as ssm from aws_cdk import core as cdk class VpcLookupUtility(cdk.Construct): def __init__(self, scope: cdk.Construct, construct_id: str, env_name: str, **kwargs) -> None: super().__init__(scope, construct_id, **kwargs) # 从SSM按环境路径读取参数(假设参数路径格式为 /{env}/vpc/xxx) self.vpc_id = ssm.StringParameter.from_string_parameter_attributes( self, "VpcIdParam", parameter_name=f"/{env_name}/vpc/id" ).string_value self.vpc_cidr = ssm.StringParameter.from_string_parameter_attributes( self, "VpcCidrParam", parameter_name=f"/{env_name}/vpc/cidr" ).string_value # 子网ID、可用区、路由表ID按逗号分隔存储,转为列表 subnet_ids_str = ssm.StringParameter.from_string_parameter_attributes( self, "PrivateSubnetIdsParam", parameter_name=f"/{env_name}/vpc/private-subnet-ids" ).string_value self.subnet_list = subnet_ids_str.split(",") azs_str = ssm.StringParameter.from_string_parameter_attributes( self, "AvailabilityZonesParam", parameter_name=f"/{env_name}/vpc/availability-zones" ).string_value self.availability_zone_list = azs_str.split(",") route_table_ids_str = ssm.StringParameter.from_string_parameter_attributes( self, "RouteTableIdsParam", parameter_name=f"/{env_name}/vpc/private-route-table-ids" ).string_value self.route_table_id_list = route_table_ids_str.split(",") # 生成VPC引用对象 self.vpc = ec2.Vpc.from_vpc_attributes( self, "LookupVpc", vpc_id=self.vpc_id, vpc_cidr_block=self.vpc_cidr, availability_zones=self.availability_zone_list, private_subnet_ids=self.subnet_list, private_subnet_route_table_ids=self.route_table_id_list ) # 按需添加其他需要的属性,比如安全组ID self.vpc_api_gw_security_group_id = ssm.StringParameter.from_string_parameter_attributes( self, "ApiGwSgIdParam", parameter_name=f"/{env_name}/vpc/api-gw-sg-id" ).string_value
在其他CDK Stack中调用
在任意Stack里实例化这个工具类,直接使用它暴露的属性即可:
from aws_cdk import core as cdk from aws_cdk import aws_ecs as ecs from path.to.your.utility import VpcLookupUtility class MyApplicationStack(cdk.Stack): def __init__(self, scope: cdk.Construct, construct_id: str, env_name: str, **kwargs) -> None: super().__init__(scope, construct_id, **kwargs) # 实例化工具类,传入环境名称 vpc_util = VpcLookupUtility(self, "VpcLookup", env_name=env_name) # 使用获取到的VPC创建资源,比如ECS集群 ecs_cluster = ecs.Cluster( self, "MyCluster", vpc=vpc_util.vpc, cluster_name=f"{env_name}-ecs-cluster" ) # 直接使用其他属性,比如安全组ID api_gw_sg = ec2.SecurityGroup.from_security_group_id( self, "ApiGwSg", security_group_id=vpc_util.vpc_api_gw_security_group_id )
方案二:纯工具函数(更轻量)
如果不需要封装成Construct,也可以写一个纯函数完成VPC查找:
from aws_cdk import aws_ec2 as ec2 from aws_cdk import aws_ssm as ssm from aws_cdk import core as cdk def get_vpc_from_ssm(scope: cdk.Construct, env_name: str) -> ec2.IVpc: # 读取SSM参数 vpc_id = ssm.StringParameter.from_string_parameter_attributes( scope, f"VpcId-{env_name}", parameter_name=f"/{env_name}/vpc/id" ).string_value vpc_cidr = ssm.StringParameter.from_string_parameter_attributes( scope, f"VpcCidr-{env_name}", parameter_name=f"/{env_name}/vpc/cidr" ).string_value subnet_ids = ssm.StringParameter.from_string_parameter_attributes( scope, f"SubnetIds-{env_name}", parameter_name=f"/{env_name}/vpc/private-subnet-ids" ).string_value.split(",") azs = ssm.StringParameter.from_string_parameter_attributes( scope, f"Azs-{env_name}", parameter_name=f"/{env_name}/vpc/availability-zones" ).string_value.split(",") route_table_ids = ssm.StringParameter.from_string_parameter_attributes( scope, f"RouteTableIds-{env_name}", parameter_name=f"/{env_name}/vpc/private-route-table-ids" ).string_value.split(",") # 返回VPC引用 return ec2.Vpc.from_vpc_attributes( scope, f"LookupVpc-{env_name}", vpc_id=vpc_id, vpc_cidr_block=vpc_cidr, availability_zones=azs, private_subnet_ids=subnet_ids, private_subnet_route_table_ids=route_table_ids )
调用方式
在Stack中直接调用函数:
class MyAnotherStack(cdk.Stack): def __init__(self, scope: cdk.Construct, construct_id: str, env_name: str, **kwargs) -> None: super().__init__(scope, construct_id, **kwargs) vpc = get_vpc_from_ssm(self, env_name=env_name) # 使用vpc创建后续资源...
关键注意事项
- SSM权限:运行CDK synth/deploy的用户/角色需要具备
ssm:GetParameter权限,才能读取指定的SSM参数。 - 参数格式:确保SSM中存储的子网ID、可用区等是逗号分隔的字符串,方便直接转为列表。
- 无跨引用问题:这个工具类/函数只是引用已存在的资源,不会创建新的部署资源,因此不会产生跨Stack的依赖关系,多个仓库的Stack都可以直接引入使用。
内容的提问来源于stack exchange,提问作者Codistan
相关产品推荐
相关产品推荐

