组件化、模块化与 Automation API

ComponentResource 深入：封装可复用的 VPC 组件

组件设计原则

一个好的 ComponentResource 应该：封装一个完整的"关注点"（如一套 VPC 网络）；暴露高层配置参数（不暴露内部实现细节）；通过 getter 方法或属性暴露子资源的关键输出。

// vpc-component.ts — 可复用的 VPC 组件（TypeScript）
import * as aws from "@pulumi/aws";
import * as pulumi from "@pulumi/pulumi";

export interface VpcComponentArgs {
    cidrBlock?: pulumi.Input<string>;
    azCount?: number;
    enableNatGateway?: boolean;
    tags?: pulumi.Input<{[key: string]: string}>;
}

export class VpcComponent extends pulumi.ComponentResource {
    public readonly vpcId: pulumi.Output<string>;
    public readonly publicSubnetIds: pulumi.Output<string[]>;
    public readonly privateSubnetIds: pulumi.Output<string[]>;

    constructor(
        name: string,
        args: VpcComponentArgs = {},
        opts?: pulumi.ComponentResourceOptions
    ) {
        super("mycompany:network:Vpc", name, {}, opts);

        const {
            cidrBlock = "10.0.0.0/16",
            azCount = 2,
            enableNatGateway = true,
            tags = {},
        } = args;

        const childOpts: pulumi.ResourceOptions = { parent: this };

        // 创建 VPC
        const vpc = new aws.ec2.Vpc(`${name}-vpc`, {
            cidrBlock,
            enableDnsHostnames: true,
            enableDnsSupport: true,
            tags: { Name: name, ...tags },
        }, childOpts);

        // 查询可用区
        const azs = aws.getAvailabilityZonesOutput({ state: "available" });
        const selectedAzs = azs.names.apply(names => names.slice(0, azCount));

        // 创建子网（使用 pulumi.all 处理 Output 数组）
        const publicSubnets = selectedAzs.apply(azNames =>
            azNames.map((az, i) =>
                new aws.ec2.Subnet(`${name}-public-${i}`, {
                    vpcId: vpc.id,
                    cidrBlock: `10.0.${i}.0/24`,
                    availabilityZone: az,
                    mapPublicIpOnLaunch: true,
                    tags: { Name: `${name}-public-${i}`, Type: "public" },
                }, childOpts)
            )
        );

        const privateSubnets = selectedAzs.apply(azNames =>
            azNames.map((az, i) =>
                new aws.ec2.Subnet(`${name}-private-${i}`, {
                    vpcId: vpc.id,
                    cidrBlock: `10.0.${i + 10}.0/24`,
                    availabilityZone: az,
                    tags: { Name: `${name}-private-${i}`, Type: "private" },
                }, childOpts)
            )
        );

        // 暴露输出
        this.vpcId = vpc.id;
        this.publicSubnetIds = publicSubnets.apply(s => s.map(sub => sub.id));
        this.privateSubnetIds = privateSubnets.apply(s => s.map(sub => sub.id));

        this.registerOutputs({
            vpcId: this.vpcId,
            publicSubnetIds: this.publicSubnetIds,
            privateSubnetIds: this.privateSubnetIds,
        });
    }
}

// 使用：像普通资源一样
const network = new VpcComponent("prod-network", {
    cidrBlock: "10.0.0.0/16",
    azCount: 3,
    enableNatGateway: true,
});

export const vpcId = network.vpcId;
export const privateSubnetIds = network.privateSubnetIds;

模块跨项目复用

发布为 npm 包（TypeScript）

// package.json — 发布到私有 npm registry
{
  "name": "@mycompany/pulumi-components",
  "version": "1.2.0",
  "main": "./bin/index.js",
  "types": "./bin/index.d.ts",
  "scripts": {
    "build": "tsc",
    "prepublishOnly": "npm run build"
  },
  "peerDependencies": {
    "@pulumi/pulumi": "^3.0.0",
    "@pulumi/aws": "^6.0.0"
  }
}

# 发布到私有 npm
npm publish --registry https://npm.mycompany.com

# 在其他项目中使用
npm install @mycompany/pulumi-components

发布为 PyPI 包（Python）

# 构建并发布到私有 PyPI
pip install build twine
python -m build
twine upload --repository-url https://pypi.mycompany.com dist/*

# 在其他项目中使用
pip install mycompany-pulumi-components

Pulumi Registry：公共组件生态

Pulumi Registry（www.pulumi.com/registry）提供数百个开源 Provider 和 Component。一些常用的社区组件：

// 使用 @pulumi/awsx 一行创建完整 VPC
import * as awsx from "@pulumi/awsx";

const vpc = new awsx.ec2.Vpc("my-vpc", {
    numberOfAvailabilityZones: 3,
    natGateways: { strategy: "Single" },
});

// 使用 @pulumi/eks 快速创建 EKS 集群
import * as eks from "@pulumi/eks";

const cluster = new eks.Cluster("my-cluster", {
    vpcId: vpc.vpcId,
    publicSubnetIds: vpc.publicSubnetIds,
    privateSubnetIds: vpc.privateSubnetIds,
    instanceType: "t3.medium",
    desiredCapacity: 3,
});

export const kubeconfig = cluster.kubeconfig;

Automation API：代码中驱动 Pulumi

Automation API 是 Pulumi 最强大的特性之一：它允许你在普通 Python/TypeScript 程序中以编程方式调用 Pulumi（不需要 CLI），可以用来构建自定义部署平台、自动化测试流水线、动态扩缩容工具等。

# deploy_tool.py — 用 Automation API 构建自定义部署工具
from pulumi import automation as auto
import pulumi_aws as aws
import pulumi

def create_pulumi_program(bucket_name: str):
    """返回一个 Pulumi 程序函数"""
    def program():
        bucket = aws.s3.BucketV2(
            "my-bucket",
            bucket=bucket_name,
            tags={"ManagedBy": "AutomationAPI"},
        )
        pulumi.export("bucket_name", bucket.id)
    return program

def deploy_bucket(stack_name: str, bucket_name: str):
    """编程方式创建/更新 Stack"""

    # 创建或选择 Stack（inline program 不需要项目目录）
    stack = auto.create_or_select_stack(
        stack_name=stack_name,
        project_name="bucket-deployer",
        program=create_pulumi_program(bucket_name),
    )

    # 设置配置
    stack.set_config("aws:region", auto.ConfigValue("us-east-1"))

    # 安装 Provider（自动）
    stack.workspace.install_plugin("aws", "v6.0.0")

    # 预览变更
    preview_result = stack.preview()
    print(f"Preview: {preview_result.change_summary}")

    # 执行部署
    up_result = stack.up(on_output=print)   # on_output 实时打印日志
    print(f"Deployed! Outputs: {up_result.outputs}")

    # 读取输出值
    bucket_name_output = up_result.outputs["bucket_name"].value
    return bucket_name_output

# 示例：批量为每个租户创建独立的 S3 Bucket
tenants = ["tenant-a", "tenant-b", "tenant-c"]
for tenant in tenants:
    bucket = deploy_bucket(
        stack_name=f"storage-{tenant}",
        bucket_name=f"myapp-{tenant}-data",
    )
    print(f"Tenant {tenant} bucket: {bucket}")

Automation API 典型使用场景

本章小结