Boto:如何检查CloudFormation堆栈是否存在?

13 投票
6 回答
19606 浏览
提问于 2025-04-18 02:23

怎样才能用Boto检查一个CloudFormation堆栈是否存在,并且没有处于故障状态呢?这里的故障状态指的是失败和回滚状态。

我不想用try/except这种方式,因为Boto会把它记录为错误,而在我的情况下,这会把异常日志发送到报警系统。


目前我有以下几种解决方案:

1) 使用boto.cloudformation.connection.CloudFormationConnection.describe_stacks()

valid_states = '''\
CREATE_IN_PROGRESS
CREATE_COMPLETE
UPDATE_IN_PROGRESS
UPDATE_COMPLETE_CLEANUP_IN_PROGRESS
UPDATE_COMPLETE'''.splitlines()

def describe_stacks():
    result = []
    resp = cf_conn.describe_stacks()
    result.extend(resp)
    while resp.next_token:
        resp = cf_conn.describe_stacks(next_token=resp.next_token)
        result.extend(resp)
    return result


stacks = [stack for stack in describe_stacks() if stack.stack_name == STACK_NAME and stack.stack_status in valid_states]
exists = len(stacks) >= 1

这个方法比较慢,因为我有很多堆栈。


2) 使用boto.cloudformation.connection.CloudFormationConnection.list_stacks()

def list_stacks(filters):
    result = []
    resp = cf_conn.list_stacks(filters)
    result.extend(resp)
    while resp.next_token:
        resp = cf_conn.list_stacks(filters, next_token=resp.next_token)
        result.extend(resp)
    return result

stacks = [stack for stack in list_stacks(valid_states) if stack.stack_name == STACK_NAME]
exists = len(stacks) >= 1

这个方法也很慢,因为总结信息会保留90天,而我有很多堆栈。


问题是:有没有什么理想的解决方案来检查一个特定的堆栈是否存在,并且没有处于失败或回滚状态?

6 个回答

0

解决这个问题的最好方法是把它分成两个小问题:

  1. 找出哪些堆栈不存在。
  2. 找出哪些堆栈处于错误状态。

可能看起来像这样:

failure_states = ['CREATE_FAILED', ... ]
stack_names = ['prod', 'dev', 'test']

c = boto.cloudformation.connect_to_region(region)
existing_stacks = [s.stack_name for s in c.describe_stacks()]
nonexistent_stacks = set(stack_names) - set(existing_stacks)
error_stacks = c.list_stacks(stack_status_filters=failure_states) 

你可能有比我更多的堆栈,所以你可能需要使用 next_token 来分页,这可能会有点麻烦。不过,你可以看到这两个操作都可以通过各自的一次请求来完成,而且不会出现异常。

1

我会通过分页器和ListStacks这个API来检查,因为我的lambda函数或者我自己可能没有权限访问cloudformation。如果真是这样,我应该返回相关的信息,而不是说这个堆栈不存在。

import boto3
from botocore.exceptions import ClientError

cfn = boto3.client('cloudformation')
def stack_exists(stack_name: str, stack_status: str) -> bool:
    try:
        paginator = cfn.get_paginator('list_stacks')
        response_iterator = paginator.paginate()
        for page in response_iterator:
            for stack in page['StackSummaries']:
                if stack_name == stack.get('StackName') and stack.get('StackStatus') != stack_status:
                    return True
    except ClientError as e:
        logger.exception(f'Client error while checking stack : {e}')
        raise
    except Exception:
        logger.exception('Error while checking stack')
        raise
    return False

或者如果我们不想遍历所有的堆栈:

import boto3
from botocore.exceptions import ClientError

cfn = boto3.client('cloudformation')
def stack_exists(stack_name: str, required_status='DELETE_COMPLETE'):
    try:
        stacks_summary = cfn.describe_stacks(StackName=stack_name)
        stack_info = stacks_summary.get('Stacks')[0]
        return stack_name == stack_info.get('StackName') and stack_info.get('StackStatus') != required_status
    except ClientError as e:
        stack_not_found_error = f'Stack with id {stack_name} does not exist'
        error_received = e.response('Error')
        error_code_received = error_received.get('Code')
        error_message_received = error_received.get('Message')
        if error_code_received == 'ValidationError' and error_message_received == stack_not_found_error:
            return True
        logger.exception(f'Client error while describing stacks: {e}')
        raise
    except Exception:
        logger.exception('Error while checking stack')
        raise
1

我知道这个问题有点老了,但几周前有人问过有没有解决办法,所以我来分享一下...

如果你看一下boto3的文档,会发现里面经常提到已删除的堆栈。要做到这一点,你必须使用完整的堆栈ID,而不能仅仅使用堆栈名称。这是因为堆栈ID是唯一的,只有它能真正标识一个堆栈。

举个例子:

resource = boto3.resource('cloudformation')
status = resource.Stack('id:of:stack').stack_status

只有在这个堆栈ID不存在的情况下,才会返回一个异常。

谢谢!

4

来自boto文档:

describe_stacks(stack_name_or_id=None, next_token=None)

这个函数会返回指定堆栈的描述;如果没有指定堆栈名称,它会返回所有创建的堆栈的描述。

参数: stack_name_or_id(字符串) – 这是与堆栈相关的名称或唯一标识符。

既然你知道堆栈的名称,你可以使用 describe_stacks(stack_name_or_id=STACK_NAME)。这样可以让你更快地获取信息。

12

我实现了下面的代码,它可以正常工作:

import boto3
from botocore.exceptions import ClientError

client = boto3.client('cloudformation')

def stack_exists(name, required_status = 'CREATE_COMPLETE'):
    try:
        data = client.describe_stacks(StackName = name)
    except ClientError:
        return False
    return data['Stacks'][0]['StackStatus'] == required_status

我发现之前的解决方案都不够完整,而且用boto3也没有什么快速的方法,所以我自己写了上面的代码。

撰写回答