Pulumi 实战：用 Python 替代 HCL/YAML 管理云基础设施，3 个场景告诉你为什么值得切换

痛点

运维团队用 Terraform 管理基础设施已是标配，但随着规模增长，HCL 的局限性越来越明显：

逻辑表达力不足 — count、for_each、dynamic 嵌套三层以上可读性断崖式下跌
测试困难 — HCL 没有原生单元测试，terratest 要写 Go 代码，跨语言维护成本高
复用靠 Module — 复杂参数传递像在写配置的配置，Module 嵌套深了跟 YAML Hell 没区别
团队门槛 — 运维会 Python/Go，但不一定愿意学一门只用在 IaC 的 DSL

Pulumi 的思路很直接：用你已经会的编程语言（Python/Go/TypeScript/Java）直接写基础设施代码，享受 IDE 补全、类型检查、单元测试、包管理的完整生态。

方案

Pulumi 核心架构：

Developer Code (Python/Go/TS)
       ↓
  Pulumi Engine (状态管理 + Diff)
       ↓
  Cloud Provider API (AWS/Azure/GCP/K8s)

State 管理：默认用 Pulumi Cloud（免费 tier），也支持 S3/GCS/本地文件（类似 Terraform backend）
Provider 覆盖：AWS、Azure、GCP、Kubernetes、Cloudflare 等 100+ Provider
语言 SDK：Python (pulumi-aws)、Go、TypeScript、Java、C#

与 Terraform 对比关键差异：

维度	Terraform/OpenTofu	Pulumi
语言	HCL（DSL）	Python/Go/TS（通用语言）
测试	terratest (Go)	pytest/go test/jest（原生）
循环/条件	count/for_each	原生 for/if
包管理	Module Registry	pip/npm/go mod
状态	S3/Consul/Cloud	S3/GCS/Pulumi Cloud
学习成本	学 HCL	用已有语言技能

实操步骤

场景一：Python 创建 AWS VPC + EKS 集群

# 安装 Pulumi CLI
curl -fsSL https://get.pulumi.com | sh

# 新建项目
mkdir infra-eks && cd infra-eks
pulumi new aws-python --name infra-eks --stack dev

# 安装依赖
pip install pulumi-aws pulumi-eks

编写 __main__.py：

import pulumi
import pulumi_aws as aws
import pulumi_eks as eks

# 创建 VPC — 用 Python 循环生成多 AZ 子网
vpc = aws.ec2.Vpc("main-vpc", cidr_block="10.0.0.0/16",
    enable_dns_hostnames=True,
    tags={"Name": "eks-vpc", "Env": "dev"})

azs = ["us-west-2a", "us-west-2b", "us-west-2c"]
public_subnets = []
private_subnets = []

for i, az in enumerate(azs):
    public = aws.ec2.Subnet(f"public-{az}",
        vpc_id=vpc.id,
        cidr_block=f"10.0.{i}.0/24",
        availability_zone=az,
        map_public_ip_on_launch=True,
        tags={"Name": f"public-{az}"})
    public_subnets.append(public)

    private = aws.ec2.Subnet(f"private-{az}",
        vpc_id=vpc.id,
        cidr_block=f"10.0.{i + 10}.0/24",
        availability_zone=az,
        tags={"Name": f"private-{az}"})
    private_subnets.append(private)

# 创建 EKS 集群
cluster = eks.Cluster("dev-cluster",
    vpc_id=vpc.id,
    subnet_ids=[s.id for s in private_subnets],
    instance_type="t3.medium",
    desired_capacity=3,
    min_size=2,
    max_size=5,
    tags={"Env": "dev"})

pulumi.export("kubeconfig", cluster.kubeconfig)
pulumi.export("cluster_name", cluster.core.cluster.name)

# 预览变更（等同 terraform plan）
pulumi preview

# 部署
pulumi up --yes

场景二：用 pytest 对基础设施做单元测试

# test_infra.py
import pulumi
import pytest

class MockResource:
    """模拟 Pulumi 资源用于单元测试"""
    def __init__(self, name, props):
        self.name = name
        self.props = props

# Pulumi 提供 mocking 框架
@pulumi.runtime.test
def test_vpc_cidr_not_too_broad():
    """确保 VPC CIDR 不超过 /16"""
    import ipaddress
    cidr = "10.0.0.0/16"
    network = ipaddress.ip_network(cidr)
    assert network.prefixlen >= 16, "VPC CIDR 不应大于 /16"

@pulumi.runtime.test
def test_eks_min_nodes_for_ha():
    """确保 EKS 最少 2 节点保证高可用"""
    min_size = 2
    assert min_size >= 2, "生产环境至少 2 节点"

pytest test_infra.py -v

场景三：用 Python 函数封装可复用组件

# components/tagged_bucket.py
import pulumi
import pulumi_aws as aws
from typing import Optional

class TaggedBucket(pulumi.ComponentResource):
    """带标准化标签和加密的 S3 Bucket 组件"""

    def __init__(self, name: str, env: str,
                 versioning: bool = True,
                 opts: Optional[pulumi.ResourceOptions] = None):
        super().__init__("custom:storage:TaggedBucket", name, None, opts)

        self.bucket = aws.s3.BucketV2(f"{name}-bucket",
            tags={"Env": env, "ManagedBy": "pulumi", "Team": "platform"},
            opts=pulumi.ResourceOptions(parent=self))

        # 强制加密
        aws.s3.BucketServerSideEncryptionConfigurationV2(
            f"{name}-encryption",
            bucket=self.bucket.id,
            rules=[{"apply_server_side_encryption_by_default": {
                "sse_algorithm": "aws:kms"}}],
            opts=pulumi.ResourceOptions(parent=self))

        if versioning:
            aws.s3.BucketVersioningV2(f"{name}-versioning",
                bucket=self.bucket.id,
                versioning_configuration={"status": "Enabled"},
                opts=pulumi.ResourceOptions(parent=self))

        self.register_outputs({"bucket_name": self.bucket.bucket})

# 使用：一行创建标准化 Bucket
logs_bucket = TaggedBucket("app-logs", env="prod")
data_bucket = TaggedBucket("ml-data", env="prod", versioning=False)

避坑

坑 1：State 锁机制不同于 Terraform

Pulumi Cloud 自带并发锁，但如果用 S3 backend，需要手动配置 DynamoDB 锁表：

# 使用 S3 backend 时指定
pulumi login s3://my-state-bucket
# 锁由 Pulumi 自动管理 .pulumi/locks/，无需额外 DynamoDB
# 但多人并发操作时仍建议用 Pulumi Cloud 或加 CI 串行化

坑 2：Terraform 迁移不是一键完成

pulumi convert --from terraform 能转换 HCL 到 Python，但复杂 Module 可能需要手动调整：

# 转换现有 Terraform 项目
cd existing-tf-project
pulumi convert --from terraform --language python --out ../pulumi-project

# 导入已有资源（不会重建）
pulumi import aws:ec2/instance:Instance web-server i-0abc123def456

坑 3：Python 异步输出值（Output）踩坑

Pulumi 的 Output[T] 类似 Promise，不能直接当字符串用：

# ❌ 错误 — Output 不能直接拼接
bucket_url = "s3://" + bucket.bucket  # TypeError

# ✅ 正确 — 用 apply 或 format
bucket_url = bucket.bucket.apply(lambda name: f"s3://{name}")
# 或
bucket_url = pulumi.Output.format("s3://{0}", bucket.bucket)

总结

已有 Python/Go 技能的团队，Pulumi 上手成本远低于学 HCL，复用性和测试能力碾压 Module
中小规模（< 500 资源）切换成本低，pulumi import 可无缝接管已有资源
大规模 Terraform 存量项目，建议新模块用 Pulumi、老模块渐进迁移，不必一刀切
CI/CD 集成：Pulumi 原生支持 GitHub Actions、GitLab CI，pulumi up --yes 即可无人值守部署

核心结论：Pulumi 不是 Terraform 的替代品，而是给了你用真正编程语言管理基础设施的选择。逻辑复杂度越高，Pulumi 的优势越明显。