Бото: Как лучше всего проверить, существует ли стек CloudFormation?

Как лучше всего с помощью 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 дней, а у меня много стопок.


Вопрос: Какое идеальное решение проверить, существует ли данный стек и не находится ли он в состоянии отказа или отката?


person Hugo Tavares    schedule 11.04.2014    source источник
comment
Было бы очень хорошо, если бы list_stacks принимал список фильтров stack_names. Вы когда-нибудь находили решение этой проблемы?   -  person bdrx    schedule 04.05.2017
comment
@bdrx: Нет, я не искал и не искал другого решения с той недели, когда я опубликовал этот вопрос.   -  person Hugo Tavares    schedule 04.05.2017


Ответы (6)


Я реализовал следующее, что работает:

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, поэтому я создал вышеуказанное.

person Justin Barton    schedule 04.01.2019

Из документации Boto:

description_stacks (stack_name_or_id = None, next_token = None)

Возвращает описание указанного стека; если имя стека не было указано, возвращается описание для всех созданных стеков.

Параметры: stack_name_or_id (string) - имя или уникальный идентификатор, связанный со стеком.

Поскольку вы знаете имя стека, вы можете использовать describe_stacks(stack_name_or_id=STACK_NAME). Это должно ускорить вашу работу.

person Ben Whaley    schedule 11.04.2014
comment
Если стек не существует, возникает исключение, и, как я уже сказал, недостаточно использовать try /, за исключением того, что boto регистрирует это как ошибку, а моя система предупреждений получает уведомление, если видит журнал ошибок. Спасибо, но этого недостаточно. - person Hugo Tavares; 12.04.2014
comment
@HugoTavares. чтобы решить эту проблему, я использую waiter, чтобы дождаться stack_exists только с 1 попыткой (boto3.readthedocs.io/en/latest/reference/services/) и поймать для TooManyAttemptsError - person nnattawat; 01.08.2017
comment
На случай, если кто-то еще попытается это выяснить .. Правильная ошибка - botocore.exceptions.WaiterError в последней версии. - person Alex Spence; 05.09.2018

Лучший способ сделать это - разделить это на две отдельные задачи:

  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 для разбивки на страницы, что было бы немного хлопотно. Однако вы можете видеть, что обе операции могут быть выполнены с помощью одного запроса и не вызовут исключения.

person Liyan Chang    schedule 30.11.2014
comment
Спасибо за ответ, @ liyan-chang, но ответ на вашу первую проблему бесконечен. Мой код ничего не знает о существующих стеках, и я хочу узнать, существует ли данный стек. Как лучше всего проверить, существует ли стек ABC в CloudFormation, через Boto? - person Hugo Tavares; 01.12.2014
comment
'ABC' в [s.stack_name для s в c.describe_stacks ()] - person Liyan Chang; 02.12.2014
comment
Но это может быть медленным, если стеков много ... У меня было это решение в тексте вопроса. - person Hugo Tavares; 03.12.2014
comment
Ах. Вы находитесь в затруднительном положении. Если вы не можете перечислить все стеки, вам нужно попробовать / исключить исключение и обработать его. - person Liyan Chang; 04.12.2014

Я знаю, что это старый, но кто-то спросил, есть ли решение всего пару недель назад, так что вот ...

Если вы читаете документацию по boto3, в ней довольно часто упоминаются удаленные стеки. Для этого вам необходимо использовать полный идентификатор стека. Вы не можете использовать имя стека. Это потому, что единственное, что действительно уникально в стеке, - это идентификатор.

Пример:

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

Единственный раз, когда это вернет исключение, если этот идентификатор стека не существует.

Ваше здоровье!

person Sathed    schedule 19.05.2017

Вы можете отключить логгер 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)
person Nathan Thompson    schedule 23.08.2018

Я бы проверил с помощью пагинатора в сочетании с вызовом API ListStacks, потому что возможно, моя лямбда или у меня нет разрешений на 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
person samtoddler    schedule 27.04.2021