Boto：检查 CloudFormation 堆栈是不是存在的最佳方法是啥？

Posted 2023-03-04

【中文标题】Boto：检查 CloudFormation 堆栈是不是存在的最佳方法是啥？

【英文标题】：Boto: What is the best way to check if a CloudFormation stack is exists?Boto：检查 CloudFormation 堆栈是否存在的最佳方法是什么？

【发布时间】：2014-05-25 23:55:54

【问题描述】：

使用 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 天，而且我有很多堆栈。

---

问题：检查给定堆栈是否存在且未处于故障或回滚状态的理想解决方案是什么？

【问题讨论】：

如果 list_stacks 接受 stack_names 过滤器列表，那就太好了。你有没有找到解决这个问题的方法？

@bdrx：不，我没有，自从我发布这个问题的那一周以来，我也没有寻找其他解决方案。

【参考方案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

【参考方案2】：

我实现了以下工作：

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 的快速方法，所以我创建了上述解决方案。

【参考方案3】：

您可以通过将 Boto 记录器的级别设置为高于 ERROR 来使其静音。这将让您在没有日志记录警报的情况下捕获异常：

import boto3
import botocore.exceptions
import logging

cf = boto3.resource('cloudformation')
logging.Logger.manager.loggerDict['boto3'].setLevel(logging.CRITICAL)
logging.Logger.manager.loggerDict['botocore'].setLevel(logging.CRITICAL)
try:
    stack.Stack('foo').stack_status
except botocore.exceptions.ClientError:
    logging.info('stack doesnt exist')

logging.Logger.manager.loggerDict['boto3'].setLevel(logging.WARNING)
logging.Logger.manager.loggerDict['botocore'].setLevel(logging.WARNING)

【参考方案4】：

我知道这是旧的，但几周前有人问是否有解决方案，所以这里......

如果您阅读 boto3 文档，它会经常提到已删除的堆栈。为此，您必须使用完整的堆栈 ID。您不能使用堆栈名称。这是因为堆栈唯一真正独特的是 ID。

例子：

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

唯一会返回异常的情况是该堆栈 ID 不存在。

干杯！

【参考方案5】：

最好的办法是把它分成两个独立的问题：

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

可能看起来像这样：

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 来分页，这会有点麻烦。但是，您可以看到这两个操作都可以通过一个请求完成，并且不会引发异常。

【讨论】：

感谢@liyan-chang 的回答，但您的第一个问题的答案是无限的。我的代码对现有堆栈一无所知，我想知道给定堆栈是否存在。通过 Boto 检查 CloudFormation 中是否存在堆栈“ABC”的最佳方法是什么？

'ABC' in [s.stack_name for s in c.describe_stacks()]

但是如果有很多堆栈，这可能会很慢......我在问题文本中有这个解决方案。

啊。你在一块岩石和一个艰难的地方。如果你不能列出所有的堆栈，那么你必须尝试/排除异常并处理它。

【参考方案6】：

来自 boto 文档：

describe_stacks(stack_name_or_id=None, next_token=None)

返回指定堆栈的描述；如果未指定堆栈名称，则返回所有已创建堆栈的描述。

参数：stack_name_or_id (string) – 与堆栈关联的名称或唯一标识符。

由于您知道堆栈名称，因此可以使用describe_stacks(stack_name_or_id=STACK_NAME)。这应该会为您加快速度。

【讨论】：

如果堆栈不存在，它会引发异常，正如我所说，使用 try/except 是不够的，因为 boto 将其记录为错误，如果出现错误，我的警报系统会收到通知它看到一个错误日志。谢谢，但这还不够。

@HugoTavares。为了解决这个问题，我使用 waiter 等待 stack_exists 仅尝试 1 次 (boto3.readthedocs.io/en/latest/reference/services/…) 并捕获 TooManyAttemptsError

以防万一其他人试图弄清楚..正确的错误是最新版本中的 botocore.exceptions.WaiterError。