Pulumi报错：不支持对Output[T]调用__str__，求ECS服务配置方案

问题背景 #

在创建首个Pulumi项目时，尝试将ECS任务定义配置到ECS服务中，运行代码后触发报错：

Calling str on an Output[T] is not supported.
To get the value of an Output[T] as an Output[str] consider:

1. o.apply(lambda v => f"prefix{v}suffix")
   This function may throw in a future version of Pulumi.

尝试使用task_definition=task_definition.arn.apply(lambda arn : arn)无法解决问题。

报错根源分析 #

报错并非来自task_definition.arn的传递，而是生成容器定义时直接使用了未解析的Output值：
代码中pulumi.get_stack()返回的是Output[str]类型，你直接将其赋值给task_definitions['environment']，随后用Jinja2模板渲染时，会尝试将Output对象转为字符串，触发了Pulumi的Output类型限制。

解决方案 #

需要通过apply方法解析pulumi.get_stack()的实际值，在回调中完成模板渲染和资源创建，具体修改如下：

1. 修改create_task_def函数 #

将依赖Output值的逻辑封装到apply回调中，确保只有拿到实际栈名称后再渲染模板、创建任务定义：

def create_task_def(ecs_service_name, custom_task_definitions):
    container_port_app = data['service'][ecs_service_name]['container_port_app'] if 'container_port_app' in \
        data['service'][ecs_service_name] else data['service']['default']['container_port_app']

    # 先构造不含Output的基础配置
    base_task_definitions = {
        'container_name': data['task_def'][ecs_service_name],
        'container_port_app': container_port_app,
        'image': data['task_def'][ecs_service_name]['image'],
        'is_pinned': data['service'][ecs_service_name]['is_pinned'] if 'is_pinned' in \
            data['service'][ecs_service_name] else data['service']['default']['is_pinned'],
        'network_mode': data['service'][ecs_service_name]['network_mode'] if 'network_mode' in \
            data['service'][ecs_service_name] else data['service']['default']['network_mode'],
        'log_group_name': data['task_def'][ecs_service_name]['log_group_name'] if 'log_group_name' \
            in data['task_def'][ecs_service_name] else ecs_service_name,
        'aws_region': data['service'][ecs_service_name]['aws_region'] if 'aws_region' in \
            data['service'][ecs_service_name] else data['service']['default']['aws_region'],
        'stream_prefix': data['service'][ecs_service_name]['stream_prefix'] if 'stream_prefix' in \
            data['service'][ecs_service_name] else data['service']['default']['stream_prefix'],
        'lb_target_group_port': container_port_app,
        'container_port_db': data['task_def'][ecs_service_name]['container_port_db'] if 'container_port_db' in \
            data['task_def'][ecs_service_name] else '',
        'catalina_opts': data['service'][ecs_service_name]['catalina_opts'] if 'catalina_opts' in \
            data['service'][ecs_service_name] else data['service']['default']['catalina_opts']
    }

    base_task_definitions.update(custom_task_definitions)

    # 使用apply解析pulumi.get_stack()的值，在回调中创建任务定义
    def create_task_with_stack(stack_name):
        task_definitions = base_task_definitions.copy()
        task_definitions['environment'] = stack_name

        with open('container_definitions.j2') as f:
            template = Template(f.read())
            task_def_data = template.render(**task_definitions)

        json_data = json.loads(task_def_data)

        return aws.ecs.TaskDefinition(f"{ecs_service_name}-task",
                                     execution_role_arn=aws.iam.get_role(name="ecsTaskExecutionRole").arn,
                                     family=f"{ecs_service_name}-service",
                                     container_definitions=json.dumps([json_data]),
                                     task_role_arn=aws.iam.get_role(name="ecsTaskRole").arn
                                     )

    return pulumi.get_stack().apply(create_task_with_stack)

2. 修改create_resource函数中的服务创建逻辑 #

由于create_task_def现在返回的是Output[TaskDefinition]，需要通过apply解析后再创建ECS服务：

def create_resource(ecs_service_name, custom_task_definitions):

    current = aws.get_caller_identity()
    pulumi.export("accountId", current.account_id)

    # Get VPC
    vpc = aws.ec2.get_vpcs()
    vpc_id = vpc.ids[1]
    subnets = aws.ec2.get_subnets(filters=[aws.ec2.GetSubnetFilterArgs(name="vpc-id", values=[vpc_id])]).ids

    sg_names = data['service'][ecs_service_name]['security_groups']
    sg_ids = sg_ids_getter.get_sg_id(sg_names, vpc_id)

    ecs_cluster_name = data['service'][ecs_service_name]['cluster_name']
    ecs_cluster = aws.ecs.Cluster.get(ecs_cluster_name, ecs_cluster_name)

    # 现在task_definition是Output[TaskDefinition]类型
    task_definition_output = create_task_def(ecs_service_name, custom_task_definitions)

    enable_ecs_managed_tags = data['service'][ecs_service_name]['enable_ecs_managed_tags'] if \
        'enable_ecs_managed_tags' in data['service'][ecs_service_name] else \
        data['service']['default']['enable_ecs_managed_tags']
    enable_execute_command = data['service'][ecs_service_name]['enable_execute_command'] if \
        'enable_exec' in data['service'][ecs_service_name] else \
        data['service']['default']['enable_execute_command']


    target_group_name = data['service'][ecs_service_name]['target_group_name']
    target_group = aws.lb.get_target_group(name=target_group_name)

    container_port_app = data['service'][ecs_service_name]['container_port_app'] if 'container_port_app' in \
        data['service'][ecs_service_name] else data['service']['default']['container_port_app']
    container_name = data['task_def'][ecs_service_name]

    # 解析TaskDefinition后创建ECS服务
    def create_service(task_def):
        return aws.ecs.Service(ecs_service_name,
                        cluster=ecs_cluster.arn,
                        network_configuration=aws.ecs.ServiceNetworkConfigurationArgs(
                            subnets=subnets,
                            security_groups=sg_ids
                        ),
                        enable_ecs_managed_tags=enable_ecs_managed_tags,
                        enable_execute_command=enable_execute_command,
                        desired_count=data['service'][ecs_service_name]['desired_count'] if \
                            'desired_count' in data['service'][ecs_service_name] else \
                            data['service']['default']['desired_count'],
                        load_balancers=[aws.ecs.ServiceLoadBalancerArgs(
                            target_group_arn=target_group.arn,
                            container_name=container_name,
                            container_port=container_port_app,
                        )],
                        task_definition=task_def.arn
                        )

    task_definition_output.apply(create_service)

额外注意点 #

• 修改了TaskDefinition的名称为f"{ecs_service_name}-task"，避免多次调用函数时出现资源名称冲突
• 所有依赖Output值的操作都放到apply回调中，确保只在获取到实际值后再执行字符串渲染或资源创建
• 不需要对task_definition.arn做多余的apply(lambda arn: arn)，Pulumi会自动处理Output类型的参数传递

内容的提问来源于stack exchange，提问作者Alexander