Reputation: 30374
What is the best way to check if a CloudFormation stack exists and is not in a broken state using Boto? By broken I mean failed and rollback states.
I don't want to use a try/except
solution, because boto logs it as an error, and in my scenario it's going to send the exception log to an alarm system.
At the moment I have the following solutions:
1) Use 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
This is slow because I have lots of stacks.
2) Use 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
This takes forever, because summaries are kept for 90 days and I have lots of stacks.
Question: What is the ideal solution to check if a given stack exists and is not in a failure or rollback state?
Upvotes: 13
Views: 20214
Reputation: 9605
I would check via the paginator combined with ListStacks API call because it might be possible my lambda or I doesn't have permissions for cloudformation
if that happens, I should return relevant information instead of saying stack doesn't exist.
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
Or if we don't want to iterate over all the stacks:
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
Upvotes: 1
Reputation: 131
I implemented the following which works:
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
I didn't find any of the previous solutions complete, nor is there any quick way of doing it with boto3 so I created the above.
Upvotes: 13
Reputation: 335
You can silence the Boto logger by setting its level above ERROR. This will let you capture the exception without the logging alarm:
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)
Upvotes: 0
Reputation: 846
I know this is old, but someone asked if there was a solution to it just a couple of weeks ago, so here goes...
If you read the boto3 docs, it mentions deleted stacks quite frequently. In order to do this, you have to use the full stack ID. You can't use the stack name. This is because the only thing truly unique about the stack is the ID.
Example:
resource = boto3.resource('cloudformation')
status = resource.Stack('id:of:stack').stack_status
The only time this will return an exception is if that stack ID doesn't exist.
Cheers!
Upvotes: 1
Reputation: 8051
The best way to do this will be to split this into two separate problems:
Might look something like this:
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)
You may have many more stacks then I do so maybe you need to use next_token
to paginate, which would be a bit of a hassle. However, you can see that both operations can be done with a single request each and will not throw an exception.
Upvotes: 0
Reputation: 34406
From the boto docs:
describe_stacks(stack_name_or_id=None, next_token=None)
Returns the description for the specified stack; if no stack name was specified, then it returns the description for all the stacks created.
Parameters: stack_name_or_id (string) – The name or the unique identifier associated with the stack.
Since you know the stack name, you can use describe_stacks(stack_name_or_id=STACK_NAME)
. That should speed things up for you.
Upvotes: 4