Pulumi คืออะไรและต่างจาก Terraform อย่างไร
Pulumi เป็น Infrastructure as Code (IaC) platform ที่ให้เขียน infrastructure ด้วยภาษาโปรแกรมทั่วไป เช่น Python, TypeScript, Go, C# แทนที่จะใช้ domain-specific language (DSL) เหมือน Terraform HCL ทำให้ใช้ loops, conditions, functions, classes และ packages ที่คุ้นเคยได้เลย
ข้อดีของ Pulumi เทียบกับ Terraform ได้แก่ ใช้ภาษาโปรแกรมจริง (ไม่ต้องเรียน HCL), IDE support ดีเยี่ยม (autocomplete, type checking, refactoring), reusable components ด้วย classes และ packages, testing ด้วย testing frameworks ปกติ (pytest, jest), Automation API สำหรับ embed IaC ใน applications และ state management ผ่าน Pulumi Cloud หรือ self-hosted backends
Pulumi รองรับ cloud providers หลักทุกตัว ได้แก่ AWS, Azure, Google Cloud, Kubernetes, Docker, DigitalOcean และอีกมากมาย สามารถ import existing resources จาก Terraform state หรือ cloud provider ได้
Automation Scripts ใน Pulumi context หมายถึงการใช้ Pulumi Automation API เพื่อสร้าง infrastructure programmatically โดยไม่ต้องใช้ CLI ใช้สำหรับ self-service platforms, custom deployment tools, integration testing และ dynamic infrastructure provisioning
ติดตั้ง Pulumi และเริ่มต้นใช้งาน
วิธีติดตั้งและสร้าง project แรก
# === ติดตั้ง Pulumi ===
# Linux/macOS
curl -fsSL https://get.pulumi.com | sh
# Windows (PowerShell)
# iwr https://get.pulumi.com/install.ps1 -UseBasicParsing | iex
# Verify
pulumi version
# === Login to Backend ===
# Option 1: Pulumi Cloud (free tier)
pulumi login
# Option 2: Local filesystem
pulumi login --local
# Option 3: S3 backend
pulumi login s3://my-pulumi-state-bucket
# === สร้าง Project ===
mkdir my-infra && cd my-infra
pulumi new python -y --name my-infra --description "Infrastructure automation"
# Project structure:
# my-infra/
# ├── __main__.py # Main infrastructure code
# ├── Pulumi.yaml # Project configuration
# ├── Pulumi.dev.yaml # Stack-specific config
# ├── requirements.txt # Python dependencies
# └── venv/ # Virtual environment
# === ตั้งค่า Cloud Provider ===
# AWS
pip install pulumi-aws
export AWS_ACCESS_KEY_ID=your-key
export AWS_SECRET_ACCESS_KEY=your-secret
pulumi config set aws:region ap-southeast-1
# Google Cloud
pip install pulumi-gcp
export GOOGLE_CREDENTIALS=$(cat service-account.json)
pulumi config set gcp:project my-project
pulumi config set gcp:region asia-southeast1
# Kubernetes
pip install pulumi-kubernetes
export KUBECONFIG=~/.kube/config
# === Stack Management ===
pulumi stack init dev
pulumi stack init staging
pulumi stack init production
# Switch stacks
pulumi stack select dev
# Set secrets
pulumi config set --secret db_password MySecretPass123
# Preview changes
pulumi preview
# Deploy
pulumi up
# Destroy
pulumi destroy
echo "Pulumi setup complete"สร้าง Infrastructure ด้วย Python
ตัวอย่าง infrastructure code
#!/usr/bin/env python3
# __main__.py — Pulumi Infrastructure with Python
import pulumi
import pulumi_aws as aws
import json
from dataclasses import dataclass
from typing import Optional
# === Configuration ===
config = pulumi.Config()
env = pulumi.get_stack() # dev, staging, production
project = pulumi.get_project()
# === VPC and Networking ===
vpc = aws.ec2.Vpc(
f"{project}-vpc",
cidr_block="10.0.0.0/16",
enable_dns_hostnames=True,
enable_dns_support=True,
tags={"Name": f"{project}-{env}-vpc", "Environment": env},
)
# Create subnets dynamically
azs = ["ap-southeast-1a", "ap-southeast-1b", "ap-southeast-1c"]
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 * 2}.0/24",
availability_zone=az,
map_public_ip_on_launch=True,
tags={"Name": f"public-{az}", "Type": "public"},
)
public_subnets.append(public)
private = aws.ec2.Subnet(
f"private-{az}",
vpc_id=vpc.id,
cidr_block=f"10.0.{i * 2 + 1}.0/24",
availability_zone=az,
tags={"Name": f"private-{az}", "Type": "private"},
)
private_subnets.append(private)
# Internet Gateway
igw = aws.ec2.InternetGateway(
f"{project}-igw",
vpc_id=vpc.id,
tags={"Name": f"{project}-{env}-igw"},
)
# === Security Groups ===
web_sg = aws.ec2.SecurityGroup(
"web-sg",
vpc_id=vpc.id,
description="Web server security group",
ingress=[
{"protocol": "tcp", "from_port": 80, "to_port": 80, "cidr_blocks": ["0.0.0.0/0"]},
{"protocol": "tcp", "from_port": 443, "to_port": 443, "cidr_blocks": ["0.0.0.0/0"]},
],
egress=[
{"protocol": "-1", "from_port": 0, "to_port": 0, "cidr_blocks": ["0.0.0.0/0"]},
],
tags={"Name": "web-sg"},
)
db_sg = aws.ec2.SecurityGroup(
"db-sg",
vpc_id=vpc.id,
description="Database security group",
ingress=[
{"protocol": "tcp", "from_port": 5432, "to_port": 5432,
"security_groups": [web_sg.id]},
],
tags={"Name": "db-sg"},
)
# === RDS Database ===
db_subnet_group = aws.rds.SubnetGroup(
"db-subnets",
subnet_ids=[s.id for s in private_subnets],
tags={"Name": f"{project}-db-subnets"},
)
db_password = config.require_secret("db_password")
db = aws.rds.Instance(
f"{project}-db",
engine="postgres",
engine_version="16.1",
instance_class="db.t3.micro" if env == "dev" else "db.r6g.large",
allocated_storage=20,
db_name=project.replace("-", "_"),
username="admin",
password=db_password,
vpc_security_group_ids=[db_sg.id],
db_subnet_group_name=db_subnet_group.name,
skip_final_snapshot=env == "dev",
tags={"Environment": env},
)
# === S3 Bucket ===
bucket = aws.s3.Bucket(
f"{project}-assets",
acl="private",
versioning={"enabled": True},
server_side_encryption_configuration={
"rule": {
"apply_server_side_encryption_by_default": {
"sse_algorithm": "AES256",
},
},
},
tags={"Environment": env},
)
# === Outputs ===
pulumi.export("vpc_id", vpc.id)
pulumi.export("public_subnet_ids", [s.id for s in public_subnets])
pulumi.export("db_endpoint", db.endpoint)
pulumi.export("db_name", db.db_name)
pulumi.export("bucket_name", bucket.bucket)
pulumi.export("bucket_arn", bucket.arn)Automation API สำหรับ Self-Service
ใช้ Automation API สร้าง infrastructure programmatically
#!/usr/bin/env python3
# automation_api.py — Pulumi Automation API
import pulumi
from pulumi import automation as auto
import pulumi_aws as aws
import json
import logging
from typing import Dict, Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("automation")
def create_web_stack(name: str, instance_type: str = "t3.micro",
region: str = "ap-southeast-1"):
"""Programmatically create a web server stack"""
def pulumi_program():
# AMI lookup
ami = aws.ec2.get_ami(
most_recent=True,
owners=["099720109477"], # Canonical
filters=[{
"name": "name",
"values": ["ubuntu/images/hvm-ssd/ubuntu-22.04-amd64-server-*"],
}],
)
# Security group
sg = aws.ec2.SecurityGroup(
f"{name}-sg",
ingress=[
{"protocol": "tcp", "from_port": 22, "to_port": 22,
"cidr_blocks": ["0.0.0.0/0"]},
{"protocol": "tcp", "from_port": 80, "to_port": 80,
"cidr_blocks": ["0.0.0.0/0"]},
{"protocol": "tcp", "from_port": 443, "to_port": 443,
"cidr_blocks": ["0.0.0.0/0"]},
],
egress=[
{"protocol": "-1", "from_port": 0, "to_port": 0,
"cidr_blocks": ["0.0.0.0/0"]},
],
)
# User data script
user_data = """#!/bin/bash
apt-get update -y
apt-get install -y nginx
systemctl enable nginx
systemctl start nginx
echo "" > /var/www/html/index.html
"""
# EC2 instance
server = aws.ec2.Instance(
f"{name}-server",
instance_type=instance_type,
ami=ami.id,
vpc_security_group_ids=[sg.id],
user_data=user_data,
tags={"Name": name, "ManagedBy": "pulumi-automation"},
)
pulumi.export("instance_id", server.id)
pulumi.export("public_ip", server.public_ip)
pulumi.export("public_dns", server.public_dns)
return pulumi_program
class InfraManager:
def __init__(self, project_name="auto-infra", backend_url=None):
self.project_name = project_name
self.backend_url = backend_url
def _get_stack(self, stack_name, program):
stack_args = auto.InlineProgramArgs(
project_name=self.project_name,
stack_name=stack_name,
program=program,
)
stack = auto.create_or_select_stack(stack_args)
stack.set_config("aws:region", auto.ConfigValue(value="ap-southeast-1"))
return stack
def deploy(self, stack_name, program, **kwargs):
logger.info(f"Deploying stack: {stack_name}")
stack = self._get_stack(stack_name, program)
# Preview first
preview = stack.preview()
logger.info(f"Preview: {preview.change_summary}")
# Deploy
result = stack.up(on_output=lambda msg: logger.info(msg))
outputs = {k: v.value for k, v in result.outputs.items()}
logger.info(f"Deployed: {json.dumps(outputs, indent=2)}")
return {
"stack": stack_name,
"outputs": outputs,
"summary": result.summary,
}
def destroy(self, stack_name, program):
logger.info(f"Destroying stack: {stack_name}")
stack = self._get_stack(stack_name, program)
stack.destroy(on_output=lambda msg: logger.info(msg))
stack.workspace.remove_stack(stack_name)
logger.info(f"Stack {stack_name} destroyed")
def get_outputs(self, stack_name, program):
stack = self._get_stack(stack_name, program)
outputs = stack.outputs()
return {k: v.value for k, v in outputs.items()}
def list_stacks(self):
ws = auto.LocalWorkspace(project_settings=auto.ProjectSettings(
name=self.project_name,
runtime="python",
))
return ws.list_stacks()
# Usage
# manager = InfraManager()
# program = create_web_stack("my-web", instance_type="t3.small")
# result = manager.deploy("dev-web", program)
# print(f"Server IP: {result['outputs']['public_ip']}")Testing Infrastructure Code
ทดสอบ infrastructure code ด้วย pytest
#!/usr/bin/env python3
# test_infra.py — Infrastructure Tests with Pulumi
import pytest
import pulumi
import json
# === Unit Testing with Mocks ===
class PulumiMocks(pulumi.runtime.Mocks):
def new_resource(self, args):
outputs = args.inputs
if args.typ == "aws:ec2/instance:Instance":
outputs = {
**args.inputs,
"id": "i-1234567890abcdef0",
"public_ip": "54.123.45.67",
"public_dns": "ec2-54-123-45-67.compute.amazonaws.com",
}
elif args.typ == "aws:ec2/securityGroup:SecurityGroup":
outputs = {**args.inputs, "id": "sg-12345678"}
elif args.typ == "aws:s3/bucket:Bucket":
outputs = {
**args.inputs,
"id": "my-bucket",
"arn": "arn:aws:s3:::my-bucket",
"bucket": "my-bucket",
}
elif args.typ == "aws:rds/instance:Instance":
outputs = {
**args.inputs,
"id": "my-db",
"endpoint": "my-db.cluster-abc123.ap-southeast-1.rds.amazonaws.com:5432",
}
return [f"{args.name}-id", outputs]
def call(self, args):
if args.token == "aws:ec2/getAmi:getAmi":
return {"id": "ami-12345678", "architecture": "x86_64"}
return {}
pulumi.runtime.set_mocks(PulumiMocks())
# Import infrastructure AFTER setting mocks
# from __main__ import vpc, web_sg, db, bucket
# === Tests ===
@pulumi.runtime.test
def test_vpc_has_correct_cidr():
"""VPC should use 10.0.0.0/16 CIDR block"""
def check_cidr(cidr):
assert cidr == "10.0.0.0/16", f"Expected 10.0.0.0/16, got {cidr}"
# vpc.cidr_block.apply(check_cidr)
@pulumi.runtime.test
def test_security_group_allows_http():
"""Web security group should allow HTTP (port 80)"""
def check_ingress(ingress):
http_rules = [r for r in ingress if r.get("from_port") == 80]
assert len(http_rules) > 0, "No HTTP ingress rule found"
assert http_rules[0]["cidr_blocks"] == ["0.0.0.0/0"]
# web_sg.ingress.apply(check_ingress)
@pulumi.runtime.test
def test_db_not_publicly_accessible():
"""Database should not be publicly accessible"""
def check_public(public):
assert not public, "Database should not be publicly accessible"
# db.publicly_accessible.apply(check_public)
@pulumi.runtime.test
def test_bucket_has_encryption():
"""S3 bucket should have server-side encryption"""
def check_encryption(config):
assert config is not None, "Bucket must have encryption configuration"
# bucket.server_side_encryption_configuration.apply(check_encryption)
@pulumi.runtime.test
def test_bucket_has_versioning():
"""S3 bucket should have versioning enabled"""
def check_versioning(versioning):
assert versioning.get("enabled") == True
# bucket.versioning.apply(check_versioning)
# === Integration Testing ===
def test_stack_outputs():
"""Test that stack produces expected outputs"""
from pulumi import automation as auto
stack = auto.create_or_select_stack(
stack_name="test",
project_name="test-infra",
program=lambda: None, # minimal program
)
# Verify stack exists
stacks = stack.workspace.list_stacks()
assert any(s.name == "test" for s in stacks)
# === Policy Tests (CrossGuard) ===
# policy.py
# from pulumi_policy import (
# EnforcementLevel, PolicyPack, ResourceValidationPolicy,
# )
#
# def no_public_s3(args, report_violation):
# if args.resource_type == "aws:s3/bucket:Bucket":
# acl = args.props.get("acl", "")
# if acl == "public-read" or acl == "public-read-write":
# report_violation("S3 buckets must not be publicly accessible")
#
# PolicyPack("security", policies=[
# ResourceValidationPolicy(
# name="no-public-s3",
# description="Prevent public S3 buckets",
# validate=no_public_s3,
# enforcement_level=EnforcementLevel.MANDATORY,
# ),
# ])
# Run tests: pytest test_infra.py -v
# Run policy: pulumi preview --policy-pack ./policyCI/CD Pipeline สำหรับ IaC
Automate deployment ด้วย CI/CD
# === GitHub Actions CI/CD for Pulumi ===
# .github/workflows/pulumi.yml
# name: Pulumi Infrastructure
# on:
# push:
# branches: [main]
# pull_request:
# branches: [main]
#
# env:
# PULUMI_ACCESS_TOKEN: }
# AWS_ACCESS_KEY_ID: }
# AWS_SECRET_ACCESS_KEY: }
#
# jobs:
# preview:
# if: github.event_name == 'pull_request'
# runs-on: ubuntu-latest
# steps:
# - uses: actions/checkout@v4
# - uses: actions/setup-python@v5
# with:
# python-version: '3.11'
#
# - name: Install dependencies
# run: |
# pip install -r requirements.txt
# pip install pytest
#
# - name: Run tests
# run: pytest test_infra.py -v
#
# - uses: pulumi/actions@v5
# with:
# command: preview
# stack-name: dev
# comment-on-pr: true
#
# deploy:
# if: github.ref == 'refs/heads/main'
# runs-on: ubuntu-latest
# steps:
# - uses: actions/checkout@v4
# - uses: actions/setup-python@v5
# with:
# python-version: '3.11'
#
# - run: pip install -r requirements.txt
#
# - uses: pulumi/actions@v5
# with:
# command: up
# stack-name: dev
# === Makefile for Local Development ===
# Makefile
.PHONY: preview deploy destroy test lint
preview:
pulumi preview --stack dev --diff
deploy:
pulumi up --stack dev --yes
destroy:
pulumi destroy --stack dev --yes
test:
pytest test_infra.py -v
lint:
ruff check .
mypy __main__.py
outputs:
pulumi stack output --json --stack dev
refresh:
pulumi refresh --stack dev --yes
import-resource:
pulumi import aws:ec2/instance:Instance my-server i-1234567890abcdef0
# === Multi-Environment Deployment Script ===
#!/bin/bash
# deploy_all.sh
set -euo pipefail
ENVIRONMENTS=("dev" "staging" "production")
for env in ""; do
echo "=== Deploying to $env ==="
pulumi stack select "$env"
# Preview first
pulumi preview --diff
if [ "$env" = "production" ]; then
echo "Production deployment requires manual approval"
read -p "Continue? (y/n) " -n 1 -r
echo
if [[ ! $REPLY =~ ^[Yy]$ ]]; then
echo "Skipping production"
continue
fi
fi
pulumi up --yes
echo "=== $env deployed ==="
pulumi stack output --json
done
echo "All environments deployed"FAQ คำถามที่พบบ่อย
Q: Pulumi กับ Terraform ควรเลือกตัวไหน?
A: เลือก Pulumi เมื่อ ทีมเป็น developers ที่คุ้นเคย Python/TypeScript/Go, ต้องการ complex logic ใน infrastructure code (loops, conditions), ต้องการ Automation API สำหรับ self-service platforms, ต้องการ testing ด้วย standard testing frameworks เลือก Terraform เมื่อ ทีมคุ้นเคย HCL อยู่แล้ว, ต้องการ ecosystem modules ขนาดใหญ่, organization มี Terraform expertise และ tooling, ต้องการ stability จาก mature tool
Q: Pulumi state จัดเก็บที่ไหน?
A: มี 3 options หลัก Pulumi Cloud (default) ใช้ง่าย มี UI collaboration features ฟรี สำหรับ individual use S3/GCS/Azure Blob ใช้ cloud storage เป็น backend เหมาะสำหรับ organizations ที่ต้องการ control data location Local filesystem เหมาะสำหรับ development แต่ไม่แนะนำ production เพราะ state อาจหาย สำหรับ production แนะนำ Pulumi Cloud หรือ S3 backend พร้อม encryption
Q: Migrate จาก Terraform มา Pulumi ยากไหม?
A: ไม่ยากมาก Pulumi มี tf2pulumi tool แปลง HCL เป็น Pulumi code อัตโนมัติ และ pulumi import command สำหรับ import existing resources เข้า Pulumi state แนะนำ migrate ทีละ project เริ่มจาก non-critical environments ก่อน ทั้งสอง tools สามารถอยู่ร่วมกันได้ (Pulumi manage บาง resources, Terraform manage บาง resources)
Q: Automation API ใช้ตอนไหน?
A: ใช้เมื่อต้องการ embed infrastructure provisioning ใน application เช่น self-service developer portal ที่ developers สร้าง environments เอง, SaaS platform ที่ provision infrastructure per tenant, integration testing ที่สร้าง/ทำลาย infrastructure อัตโนมัติ, custom CLI tools สำหรับ infrastructure management และ chatops ที่สร้าง infrastructure ผ่าน Slack commands
