使用 boto3 列出具有特定应用程序标签的自动缩放组名称
List out auto scaling group names with a specific application tag using boto3
我试图获取应用程序标签值为 'CCC' 的自动缩放组。
名单如下,
gweb
prd-dcc-eap-w2
gweb
prd-dcc-emc
gweb
prd-dcc-ems
CCC
dev-ccc-wer
CCC
dev-ccc-gbg
CCC
dev-ccc-wer
我在下面编写的脚本给出的输出包括一个没有 CCC 标签的 ASG。
#!/usr/bin/python
import boto3
client = boto3.client('autoscaling',region_name='us-west-2')
response = client.describe_auto_scaling_groups()
ccc_asg = []
all_asg = response['AutoScalingGroups']
for i in range(len(all_asg)):
all_tags = all_asg[i]['Tags']
for j in range(len(all_tags)):
if all_tags[j]['Key'] == 'Name':
asg_name = all_tags[j]['Value']
# print asg_name
if all_tags[j]['Key'] == 'Application':
app = all_tags[j]['Value']
# print app
if all_tags[j]['Value'] == 'CCC':
ccc_asg.append(asg_name)
print ccc_asg
我得到的输出如下,
['prd-dcc-ein-w2', 'dev-ccc-hap', 'dev-ccc-wfd', 'dev-ccc-sdf']
其中 'prd-dcc-ein-w2'
是具有不同标签 'gweb'
的 asg。 CCC ASG 列表中的最后一个 (dev-ccc-msp-agt-asg)
丢失了。我需要如下输出,
dev-ccc-hap-sdf
dev-ccc-hap-gfh
dev-ccc-hap-tyu
dev-ccc-mso-hjk
我是不是漏掉了什么?
我用下面的脚本让它工作。
#!/usr/bin/python
import boto3
client = boto3.client('autoscaling',region_name='us-west-2')
response = client.describe_auto_scaling_groups()
ccp_asg = []
all_asg = response['AutoScalingGroups']
for i in range(len(all_asg)):
all_tags = all_asg[i]['Tags']
app = False
asg_name = ''
for j in range(len(all_tags)):
if 'Application' in all_tags[j]['Key'] and all_tags[j]['Value'] in ('CCP'):
app = True
if app:
if 'Name' in all_tags[j]['Key']:
asg_name = all_tags[j]['Value']
ccp_asg.append(asg_name)
print ccp_asg
如有任何疑问,欢迎随时提问。
在 boto3 中,您可以使用 Paginators with JMESPath filtering 以更简洁的方式非常有效地执行此操作。
来自 boto3 文档:
JMESPath is a query language for JSON that can be used directly on
paginated results. You can filter results client-side using JMESPath
expressions that are applied to each page of results through the
search method of a PageIterator.
When filtering with JMESPath expressions, each page of results that is
yielded by the paginator is mapped through the JMESPath expression. If
a JMESPath expression returns a single value that is not an array,
that value is yielded directly. If the result of applying the JMESPath
expression to a page of results is a list, then each value of the list
is yielded individually (essentially implementing a flat map).
下面是 Python 代码中的样子,其中提到了 Auto Scaling Group 的 Application
标签的 CCP
值:
import boto3
client = boto3.client('autoscaling')
paginator = client.get_paginator('describe_auto_scaling_groups')
page_iterator = paginator.paginate(
PaginationConfig={'PageSize': 100}
)
filtered_asgs = page_iterator.search(
'AutoScalingGroups[] | [?contains(Tags[?Key==`{}`].Value, `{}`)]'.format(
'Application', 'CCP')
)
for asg in filtered_asgs:
print asg['AutoScalingGroupName']
详细说明 Michal Gasek 的回答,这里有一个选项可以根据 tag:value 对的字典过滤 ASG。
def get_asg_name_from_tags(tags):
asg_name = None
client = boto3.client('autoscaling')
while True:
paginator = client.get_paginator('describe_auto_scaling_groups')
page_iterator = paginator.paginate(
PaginationConfig={'PageSize': 100}
)
filter = 'AutoScalingGroups[]'
for tag in tags:
filter = ('{} | [?contains(Tags[?Key==`{}`].Value, `{}`)]'.format(filter, tag, tags[tag]))
filtered_asgs = page_iterator.search(filter)
asg = filtered_asgs.next()
asg_name = asg['AutoScalingGroupName']
try:
asgX = filtered_asgs.next()
asgX_name = asg['AutoScalingGroupName']
raise AssertionError('multiple ASG\'s found for {} = {},{}'
.format(tags, asg_name, asgX_name))
except StopIteration:
break
return asg_name
例如:
asg_name = get_asg_name_from_tags({'Env':env, 'Application':'app'})
它预计只有一个结果,并通过尝试使用 next() 获得另一个结果来检查这一点。 StopIteration 是 "good" 的情况,然后它会跳出分页器循环。
执行此操作的正确方法根本不是通过 describe_auto_scaling_groups
,而是 via describe_tags
,这将允许您在服务器端进行过滤。
您可以构造一个过滤器来请求具有以下任意值的标签应用程序实例:
Filters=[
{
'Name': 'key',
'Values': [
'Application',
]
},
{
'Name': 'value',
'Values': [
'CCC',
]
},
],
然后您的结果(在响应中的 Tags
中)是匹配标签应用于自动缩放组的所有时间。您将不得不多次调用,每次调用时传回 NextToken
,以浏览所有页面的结果。
每个结果都包含应用了匹配标签的 ASG ID。一旦您获得了所有您感兴趣的 ASG ID,然后您可以调用 describe_auto_scaling_groups
来获取他们的名字。
另一种解决方案,在我看来足够简单,可以扩展:
client = boto3.client('autoscaling')
search_tags = {"environment": "stage"}
filtered_asgs = []
response = client.describe_auto_scaling_groups()
for group in response['AutoScalingGroups']:
flattened_tags = {
tag_info['Key']: tag_info['Value']
for tag_info in group['Tags']
}
if search_tags.items() <= flattened_tags.items():
filtered_asgs.append(group)
print(filtered_asgs)
我试图获取应用程序标签值为 'CCC' 的自动缩放组。
名单如下,
gweb
prd-dcc-eap-w2
gweb
prd-dcc-emc
gweb
prd-dcc-ems
CCC
dev-ccc-wer
CCC
dev-ccc-gbg
CCC
dev-ccc-wer
我在下面编写的脚本给出的输出包括一个没有 CCC 标签的 ASG。
#!/usr/bin/python
import boto3
client = boto3.client('autoscaling',region_name='us-west-2')
response = client.describe_auto_scaling_groups()
ccc_asg = []
all_asg = response['AutoScalingGroups']
for i in range(len(all_asg)):
all_tags = all_asg[i]['Tags']
for j in range(len(all_tags)):
if all_tags[j]['Key'] == 'Name':
asg_name = all_tags[j]['Value']
# print asg_name
if all_tags[j]['Key'] == 'Application':
app = all_tags[j]['Value']
# print app
if all_tags[j]['Value'] == 'CCC':
ccc_asg.append(asg_name)
print ccc_asg
我得到的输出如下,
['prd-dcc-ein-w2', 'dev-ccc-hap', 'dev-ccc-wfd', 'dev-ccc-sdf']
其中 'prd-dcc-ein-w2'
是具有不同标签 'gweb'
的 asg。 CCC ASG 列表中的最后一个 (dev-ccc-msp-agt-asg)
丢失了。我需要如下输出,
dev-ccc-hap-sdf
dev-ccc-hap-gfh
dev-ccc-hap-tyu
dev-ccc-mso-hjk
我是不是漏掉了什么?
我用下面的脚本让它工作。
#!/usr/bin/python
import boto3
client = boto3.client('autoscaling',region_name='us-west-2')
response = client.describe_auto_scaling_groups()
ccp_asg = []
all_asg = response['AutoScalingGroups']
for i in range(len(all_asg)):
all_tags = all_asg[i]['Tags']
app = False
asg_name = ''
for j in range(len(all_tags)):
if 'Application' in all_tags[j]['Key'] and all_tags[j]['Value'] in ('CCP'):
app = True
if app:
if 'Name' in all_tags[j]['Key']:
asg_name = all_tags[j]['Value']
ccp_asg.append(asg_name)
print ccp_asg
如有任何疑问,欢迎随时提问。
在 boto3 中,您可以使用 Paginators with JMESPath filtering 以更简洁的方式非常有效地执行此操作。
来自 boto3 文档:
JMESPath is a query language for JSON that can be used directly on paginated results. You can filter results client-side using JMESPath expressions that are applied to each page of results through the search method of a PageIterator.
When filtering with JMESPath expressions, each page of results that is yielded by the paginator is mapped through the JMESPath expression. If a JMESPath expression returns a single value that is not an array, that value is yielded directly. If the result of applying the JMESPath expression to a page of results is a list, then each value of the list is yielded individually (essentially implementing a flat map).
下面是 Python 代码中的样子,其中提到了 Auto Scaling Group 的 Application
标签的 CCP
值:
import boto3
client = boto3.client('autoscaling')
paginator = client.get_paginator('describe_auto_scaling_groups')
page_iterator = paginator.paginate(
PaginationConfig={'PageSize': 100}
)
filtered_asgs = page_iterator.search(
'AutoScalingGroups[] | [?contains(Tags[?Key==`{}`].Value, `{}`)]'.format(
'Application', 'CCP')
)
for asg in filtered_asgs:
print asg['AutoScalingGroupName']
详细说明 Michal Gasek 的回答,这里有一个选项可以根据 tag:value 对的字典过滤 ASG。
def get_asg_name_from_tags(tags):
asg_name = None
client = boto3.client('autoscaling')
while True:
paginator = client.get_paginator('describe_auto_scaling_groups')
page_iterator = paginator.paginate(
PaginationConfig={'PageSize': 100}
)
filter = 'AutoScalingGroups[]'
for tag in tags:
filter = ('{} | [?contains(Tags[?Key==`{}`].Value, `{}`)]'.format(filter, tag, tags[tag]))
filtered_asgs = page_iterator.search(filter)
asg = filtered_asgs.next()
asg_name = asg['AutoScalingGroupName']
try:
asgX = filtered_asgs.next()
asgX_name = asg['AutoScalingGroupName']
raise AssertionError('multiple ASG\'s found for {} = {},{}'
.format(tags, asg_name, asgX_name))
except StopIteration:
break
return asg_name
例如:
asg_name = get_asg_name_from_tags({'Env':env, 'Application':'app'})
它预计只有一个结果,并通过尝试使用 next() 获得另一个结果来检查这一点。 StopIteration 是 "good" 的情况,然后它会跳出分页器循环。
执行此操作的正确方法根本不是通过 describe_auto_scaling_groups
,而是 via describe_tags
,这将允许您在服务器端进行过滤。
您可以构造一个过滤器来请求具有以下任意值的标签应用程序实例:
Filters=[
{
'Name': 'key',
'Values': [
'Application',
]
},
{
'Name': 'value',
'Values': [
'CCC',
]
},
],
然后您的结果(在响应中的 Tags
中)是匹配标签应用于自动缩放组的所有时间。您将不得不多次调用,每次调用时传回 NextToken
,以浏览所有页面的结果。
每个结果都包含应用了匹配标签的 ASG ID。一旦您获得了所有您感兴趣的 ASG ID,然后您可以调用 describe_auto_scaling_groups
来获取他们的名字。
另一种解决方案,在我看来足够简单,可以扩展:
client = boto3.client('autoscaling')
search_tags = {"environment": "stage"}
filtered_asgs = []
response = client.describe_auto_scaling_groups()
for group in response['AutoScalingGroups']:
flattened_tags = {
tag_info['Key']: tag_info['Value']
for tag_info in group['Tags']
}
if search_tags.items() <= flattened_tags.items():
filtered_asgs.append(group)
print(filtered_asgs)