logo头像
Snippet 博客主题

AWS CloudFormation 系列--(3)Python操作CloudFormation

B站视频链接:https://www.bilibili.com/video/BV1FG411773k/
微信公众号:自刘地
自刘地

前面两篇已经对CloudFormation做了快速入门和进阶的学习,但是在创建堆栈的时候,还是在AWS控制台界面导入文本文件。这个文章的目的是介绍如何使用Python来操作CloudFormation,进一步的实现自动化操作。

一、AWS 凭证(Credentials)

1.1、凭证介绍

客户端必须拥有凭证[参考链接1]才能对AWS API进行操作,可以通过多种方式来配置凭证。配置凭证的最佳实践是,如果你在EC2上运行代码,建议使用IAM角色。如果你是用AWS开发工具包(例如Python),建议使用共享凭证文件。后面代码都是使用的共享凭证文件。

Boto3 将在搜索凭证时查找多个位置。Boto3 查找凭据的机制是搜索可能位置的列表,并在找到凭据后立即停止。Boto3 搜索凭据的顺序参考:

1
2
3
4
5
6
7
8
1.在boto.client()方法中将凭证作为参数传递
2.创建Session对象时将凭证作为参数传递
3.环境变量
4.共享凭证文件 ( ~/.aws/credentials )
5.AWS 配置文件 ( ~/.aws/config )
6.担任角色提供者
7.Boto2 配置文件(/etc/boto.cfg和~/.boto)
8.配置了 IAM 角色的 Amazon EC2 实例上的实例元数据服务。

1.2、AWS CLI配置凭证

安装AWS CLI[参考链接2]客户端之后,可以利用aws configure快速配置凭证。需要提前从AWS控制台创建对应账号的AK(Access Key)和SK(Secret Key)。

1
2
3
4
5
$ aws configure
AWS Access Key ID [None]: AKIAIOSFODNN7EXAMPLE
AWS Secret Access Key [None]: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
Default region name [None]: cn-northwest-1
Default output format [None]:

直接使用aws configure产生的是default配置文件,在没有明确指定配置文件时,就会使用default配置文件。你还可以使用aws configure --profile cn-prod来创建名称为cn-prod的配置文件。假设default配置的是测试账号信息,cn-prod配置的是生产账号信息,在不明确指定配置文件的情况下,代码会使用default配置,也就是测试环境的凭证。

配置凭证时,指定名称为cn-prod

1
2
3
4
5
$ aws configure --profile cn-prod
AWS Access Key ID [None]: AKIAI44QH8DHBEXAMPLE
AWS Secret Access Key [None]: je7MtGbClwBF/2Zp9Utk/h3yCo8nvbEXAMPLEKEY
Default region name [None]: cn-northwest-1
Default output format [None]:

AWS会将配置文件放在.aws文件夹中,这个文件夹默认在用户主目录下,例如Windows的路径C:\Users\crosswalk\.aws,Linux root用户路径/root/.aws。文件夹中会产生一个凭证配置文件credentials和一个Config配置文件config

凭证配置文件credentials中具体的内容:

1
2
3
4
5
6
7
[default]
aws_access_key_id=AKIAIOSFODNN7EXAMPLE
aws_secret_access_key=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

[cn-prod]
aws_access_key_id=AKIAI44QH8DHBEXAMPLE
aws_secret_access_key=je7MtGbClwBF/2Zp9Utk/h3yCo8nvbEXAMPLEKEY

Config配置文件config中的具体内容:

1
2
3
4
5
[default]
region=cn-northwest-1

[profile cn-prod]
region=cn-northwest-1

使用 aws configure list-profiles 命令列出所有配置文件名称。

1
2
3
C:\Users\crosswalk>aws configure list-profiles
default
cn-prod

1.3、AWS CLI凭证测试

使用默认凭证,查看测试账号下的EC2

1
2
3
4
5
6
7
8
9
10
11
C:\Users\crosswalk>aws ec2 describe-instances
{
"Reservations": [
{
"Groups": [],
"Instances": [
{
"AmiLaunchIndex": 0,
"ImageId": "ami-007315f06e322f1ab",
"InstanceId": "i-08fc1f70383191bb4",
"InstanceType": "t2.2xlarge"

指定cn-prod凭证,查看生产账号下的EC2

1
2
3
4
5
6
7
8
9
10
11
C:\Users\crosswalk>aws ec2 describe-instances --profile cn-prod
{
"Reservations": [
{
"Groups": [],
"Instances": [
{
"AmiLaunchIndex": 0,
"ImageId": "ami-7c9f8b1e",
"InstanceId": "i-099540680995ce588",
"InstanceType": "t2.medium",

1.4、boto3 切换凭证测试

你可以在创建Session[参考链接3]时通过profile_name参数指定凭证配置文件名称。

1
2
3
4
5
profile_name = 'my-global'
my_region = 'cn-northwest-1'

session = boto3.Session(profile_name=profile_name, region_name=my_region)
dev_s3_client = session.client('s3')

在创建脚本时,开头定义使用的凭证名称。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_create_stack(stackname, file_path):
"""创建堆栈"""
import boto3
session = boto3.Session(profile_name=profile_name, region_name=my_region)
client = session.client('cloudformation')
with open(f"{file_path}", 'r', encoding='UTF-8') as fd: # 读取CloudFormation YAML文件内容
templatebody = fd.read()
try:
client.create_stack(StackName=f"{stackname}", Capabilities=['CAPABILITY_IAM'], TemplateBody=templatebody)
print(f'["{stackname}"堆栈创建成功]')
except Exception as e:
print(f'["{stackname}"堆栈创建失败]')
print(e)


if __name__ == "__main__":
cft_create_stack('vpc-stack-test', 'vpc-test.yaml')

二、Python Boto3操作CloudFormation代码参考

利用Python3的Boto3模块,就可以完成AWS上的所有API操作,具体可以参考boto3 CloudFormaionl[参考链接4],查看API的详细参数。

写一个用于测试脚本的CloudFormation模板vpc-test.yaml,创建一个VPC并且输出VPC的ID信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
Resources:
MyTestVpc:
Type: AWS::EC2::VPC
Properties:
CidrBlock: 10.0.0.0/16
EnableDnsSupport: 'true'
EnableDnsHostnames: 'true'
Tags:
- Key: Name
Value: MyTestVpc

Outputs:
MyTestVpc:
Description: MyTestVpc
Value: !Ref MyTestVpc
Export:
Name: MyTestVpc-ID

2.1、创建堆栈

脚本需要指定堆栈的名称和模板的路径。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_create_stack(stackname, file_path):
"""创建堆栈"""
import boto3
session = boto3.Session(profile_name=profile_name, region_name=my_region)
client = session.client('cloudformation')
with open(f"{file_path}", 'r', encoding='UTF-8') as fd: # 读取CloudFormation YAML文件内容
templatebody = fd.read()
try:
client.create_stack(StackName=f"{stackname}", Capabilities=['CAPABILITY_IAM'], TemplateBody=templatebody)
print(f'["{stackname}"堆栈创建成功]')
except Exception as e:
print(f'["{stackname}"堆栈创建失败]')
print(e)


if __name__ == "__main__":
cft_create_stack('vpc-stack-test', 'vpc-test.yaml')

代码输出:

1
["vpc-stack-test"堆栈创建成功]

2.2、获取堆栈输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_stack_output(stackname): # 当堆栈完成后获取输出
"""获取堆栈输出内容"""
import time
import boto3
while True:
nowtime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
session = boto3.Session(profile_name=profile_name, region_name=my_region)
client = session.client('cloudformation')
response = client.describe_stacks(StackName=stackname)
if response['Stacks'][0]['StackStatus'] == 'CREATE_COMPLETE':
response = client.describe_stacks(StackName=stackname)
outputs = response["Stacks"][0]["Outputs"]
print(f'[{nowtime}] Stack Create Complete!')
print(outputs)
return outputs
else:
time.sleep(1)
print(f'[{nowtime}] Wait Stack Complete...')

if __name__ == "__main__":
# cft_create_stack('vpc-stack-test', 'vpc-test.yaml')
cfn_stack_output('vpc-stack-test')

代码输出:

1
2
[2022-04-02 11:51:26] Stack Create Complete!
[{'OutputKey': 'MyTestVpc', 'OutputValue': 'vpc-0e764b67068222644', 'Description': 'MyTestVpc', 'ExportName': 'MyTestVpc-ID'}]

2.3、删除堆栈

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
profile_name = 'cn-prod'
my_region = 'cn-northwest-1'

def cfn_delete_stack(stackname):
"""删除堆栈"""
import boto3
session = boto3.Session(profile_name=profile_name, region_name=my_region)
client = session.client('cloudformation')
try:
rsp = client.delete_stack(StackName=stackname)
print(f'["{stackname}"堆栈删除成功]')
except Exception as e:
print(f'["{stackname}"堆栈删除失败]')
print(e)

if __name__ == "__main__":
# cft_create_stack('vpc-stack-test', 'vpc-test.yaml')
# cfn_stack_output('vpc-stack-test')
cfn_delete_stack('vpc-stack-test')

代码输出:

1
["vpc-stack-test"堆栈删除成功]

三、参考文档