使用 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)